Module demo_api.utils.api

API unifiée pour demo_api

Interface fluide pour accéder aux fonctionnalités de l'API : - Api(base_url).users.get() - Api(base_url).vms.get() - Api(base_url).login(email, password) - Api(base_url).users.create_vm(…)

Sub-modules

demo_api.utils.api.auth

demo_api.utils.api.decorators

Décorateurs spécifiques aux appels API.

demo_api.utils.api.exceptions

Exceptions personnalisées pour l'API demo_api …

demo_api.utils.api.user

demo_api.utils.api.vm

Functions

def create_authenticated_client(base_url: str | None = None,
email: str | None = None,
password: str | None = None) ‑> ApiClient

Expand source code

def create_authenticated_client(
    base_url: Optional[str] = None,
    email: Optional[str] = None,
    password: Optional[str] = None,
) -> ApiClient:
    """
    Crée un client API avec authentification automatique

    Args:
        base_url: URL de base de l'API
        email: Email pour l'authentification
        password: Mot de passe pour l'authentification

    Returns:
        ApiClient: Client API authentifié
    """
    client = ApiClient(base_url)

    # Essayer d'abord avec les identifiants fournis
    if email and password:
        try:
            client.login(email, password)
            return client
        except UserLoginError:
            logger.warning(
                "Échec de connexion avec les identifiants fournis", email=email
            )

    # Essayer avec les identifiants de la configuration
    if config.has_credentials:
        try:
            client.login(config.DEMO_API_EMAIL, config.DEMO_API_PASSWORD)
            return client
        except UserLoginError:
            logger.warning("Échec de connexion avec les identifiants de configuration")

    # Si pas d'authentification, retourner le client sans token
    logger.warning("Aucune authentification réussie, client créé sans token")
    return client

Crée un client API avec authentification automatique

Args

base_url

URL de base de l'API

email

Email pour l'authentification

password

Mot de passe pour l'authentification

Returns

ApiClient

Client API authentifié

Classes

class ApiClient (base_url: str | None = None, token: str | None = None)

Expand source code

class ApiClient:
    """Client API unifié avec interface fluide"""

    def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None):
        """
        Initialise le client API

        Args:
            base_url: URL de base de l'API (utilise config par défaut)
            token: Token d'authentification (optionnel)
        """
        self.base_url = base_url or config.DEMO_API_BASE_URL
        self.token = token or config.DEMO_API_TOKEN
        self._auth = Auth(self.base_url)

        # Interfaces spécialisées
        self.users = UsersAPI(self)
        self.vms = VMsAPI(self)
        self.auth = AuthAPI(self)

        logger.info(
            "Client API unifié initialisé",
            base_url=self.base_url,
            has_token=bool(self.token),
        )

    def login(self, email: str, password: str) -> str:
        """Méthode de connexion directe (raccourci)"""
        return self.auth.login(email, password)

    def create_user(self, name: str, email: str, password: str) -> str:
        """Méthode de création d'utilisateur directe (raccourci)"""
        return self.auth.create_user(name, email, password)

    def get_user_info(self) -> Dict[str, Any]:
        """Méthode d'information utilisateur directe (raccourci)"""
        return self.auth.get_user_info()

    def is_authenticated(self) -> bool:
        """Vérifie si le client est authentifié"""
        return bool(self.token)

    def set_token(self, token: str) -> None:
        """Définit le token d'authentification"""
        self.token = token
        logger.info("Token défini pour le client API", token_length=len(token))

    def clear_token(self) -> None:
        """Supprime le token d'authentification"""
        self.token = None
        logger.info("Token supprimé du client API")

    # Méthodes de convenance pour un accès direct aux fonctionnalités principales
    def get_all_data(self) -> Dict[str, Any]:
        """Récupère toutes les données (utilisateurs et VMs) et les associe

        Returns:
            Dict contenant les utilisateurs avec leurs VMs associées

        Raises:
            VMsFetchError: Si la récupération des données échoue
        """
        logger.info("Récupération de toutes les données via API unifiée")
        users = self.users.get()
        vms = self.vms.get()
        self.users.add_vms_to_users(users, vms)
        return {
            "users": users,
            "vms": vms,
            "total_users": len(users),
            "total_vms": len(vms),
            "users_with_vms": len([u for u in users if u.get("vms")]),
        }

    def get_user_data(self, user_id: int) -> Dict[str, Any]:
        """Récupère un utilisateur avec ses VMs associées

        Args:
            user_id: ID de l'utilisateur

        Returns:
            Dict contenant l'utilisateur avec ses VMs

        Raises:
            UsersFetchError: Si l'utilisateur n'existe pas
        """
        logger.info(
            "Récupération des données d'un utilisateur via API unifiée", user_id=user_id
        )
        user = self.users.get_user(user_id)

        # Récupérer toutes les VMs et trouver celles de cet utilisateur
        all_vms = self.vms.get()
        user_vms = [vm for vm in all_vms if vm.get("user_id") == user_id]
        user["vms"] = user_vms

        return {"user": user, "vm_count": len(user_vms), "vms": user_vms}

    def get_vm_data(self, vm_id: int) -> Dict[str, Any]:
        """Récupère une VM avec les informations de son utilisateur

        Args:
            vm_id: ID de la VM

        Returns:
            Dict contenant la VM avec les infos utilisateur

        Raises:
            VMsFetchError: Si la VM n'existe pas
        """
        logger.info("Récupération des données d'une VM via API unifiée", vm_id=vm_id)
        vm = self.vms.get_vm(vm_id)

        # Récupérer les informations de l'utilisateur propriétaire
        if vm.get("user_id"):
            try:
                owner = self.users.get_user(vm["user_id"])
                vm["owner"] = owner
            except Exception:
                vm["owner"] = None

        return vm

    def __repr__(self) -> str:
        """Représentation du client API"""
        return f"ApiClient(base_url='{self.base_url}', authenticated={self.is_authenticated()})"

Client API unifié avec interface fluide

Initialise le client API

Args

base_url

URL de base de l'API (utilise config par défaut)

token

Token d'authentification (optionnel)

Methods

def clear_token(self) ‑> None

Expand source code

def clear_token(self) -> None:
    """Supprime le token d'authentification"""
    self.token = None
    logger.info("Token supprimé du client API")

Supprime le token d'authentification

def create_user(self, name: str, email: str, password: str) ‑> str

Expand source code

def create_user(self, name: str, email: str, password: str) -> str:
    """Méthode de création d'utilisateur directe (raccourci)"""
    return self.auth.create_user(name, email, password)

Méthode de création d'utilisateur directe (raccourci)

def get_all_data(self) ‑> Dict[str, Any]

Expand source code

def get_all_data(self) -> Dict[str, Any]:
    """Récupère toutes les données (utilisateurs et VMs) et les associe

    Returns:
        Dict contenant les utilisateurs avec leurs VMs associées

    Raises:
        VMsFetchError: Si la récupération des données échoue
    """
    logger.info("Récupération de toutes les données via API unifiée")
    users = self.users.get()
    vms = self.vms.get()
    self.users.add_vms_to_users(users, vms)
    return {
        "users": users,
        "vms": vms,
        "total_users": len(users),
        "total_vms": len(vms),
        "users_with_vms": len([u for u in users if u.get("vms")]),
    }

Récupère toutes les données (utilisateurs et VMs) et les associe

Returns

Dict contenant les utilisateurs avec leurs VMs associées

Raises

VMsFetchError

Si la récupération des données échoue

def get_user_data(self, user_id: int) ‑> Dict[str, Any]

Expand source code

def get_user_data(self, user_id: int) -> Dict[str, Any]:
    """Récupère un utilisateur avec ses VMs associées

    Args:
        user_id: ID de l'utilisateur

    Returns:
        Dict contenant l'utilisateur avec ses VMs

    Raises:
        UsersFetchError: Si l'utilisateur n'existe pas
    """
    logger.info(
        "Récupération des données d'un utilisateur via API unifiée", user_id=user_id
    )
    user = self.users.get_user(user_id)

    # Récupérer toutes les VMs et trouver celles de cet utilisateur
    all_vms = self.vms.get()
    user_vms = [vm for vm in all_vms if vm.get("user_id") == user_id]
    user["vms"] = user_vms

    return {"user": user, "vm_count": len(user_vms), "vms": user_vms}

Récupère un utilisateur avec ses VMs associées

Args

user_id

ID de l'utilisateur

Returns

Dict contenant l'utilisateur avec ses VMs

Raises

UsersFetchError

Si l'utilisateur n'existe pas

def get_user_info(self) ‑> Dict[str, Any]

Expand source code

def get_user_info(self) -> Dict[str, Any]:
    """Méthode d'information utilisateur directe (raccourci)"""
    return self.auth.get_user_info()

Méthode d'information utilisateur directe (raccourci)

def get_vm_data(self, vm_id: int) ‑> Dict[str, Any]

Expand source code

def get_vm_data(self, vm_id: int) -> Dict[str, Any]:
    """Récupère une VM avec les informations de son utilisateur

    Args:
        vm_id: ID de la VM

    Returns:
        Dict contenant la VM avec les infos utilisateur

    Raises:
        VMsFetchError: Si la VM n'existe pas
    """
    logger.info("Récupération des données d'une VM via API unifiée", vm_id=vm_id)
    vm = self.vms.get_vm(vm_id)

    # Récupérer les informations de l'utilisateur propriétaire
    if vm.get("user_id"):
        try:
            owner = self.users.get_user(vm["user_id"])
            vm["owner"] = owner
        except Exception:
            vm["owner"] = None

    return vm

Récupère une VM avec les informations de son utilisateur

Args

vm_id

ID de la VM

Returns

Dict contenant la VM avec les infos utilisateur

Raises

VMsFetchError

Si la VM n'existe pas

def is_authenticated(self) ‑> bool

Expand source code

def is_authenticated(self) -> bool:
    """Vérifie si le client est authentifié"""
    return bool(self.token)

Vérifie si le client est authentifié

def login(self, email: str, password: str) ‑> str

Expand source code

def login(self, email: str, password: str) -> str:
    """Méthode de connexion directe (raccourci)"""
    return self.auth.login(email, password)

Méthode de connexion directe (raccourci)

def set_token(self, token: str) ‑> None

Expand source code

def set_token(self, token: str) -> None:
    """Définit le token d'authentification"""
    self.token = token
    logger.info("Token défini pour le client API", token_length=len(token))

Définit le token d'authentification

class Api (base_url: str | None = None, token: str | None = None)

Expand source code

class ApiClient:
    """Client API unifié avec interface fluide"""

    def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None):
        """
        Initialise le client API

        Args:
            base_url: URL de base de l'API (utilise config par défaut)
            token: Token d'authentification (optionnel)
        """
        self.base_url = base_url or config.DEMO_API_BASE_URL
        self.token = token or config.DEMO_API_TOKEN
        self._auth = Auth(self.base_url)

        # Interfaces spécialisées
        self.users = UsersAPI(self)
        self.vms = VMsAPI(self)
        self.auth = AuthAPI(self)

        logger.info(
            "Client API unifié initialisé",
            base_url=self.base_url,
            has_token=bool(self.token),
        )

    def login(self, email: str, password: str) -> str:
        """Méthode de connexion directe (raccourci)"""
        return self.auth.login(email, password)

    def create_user(self, name: str, email: str, password: str) -> str:
        """Méthode de création d'utilisateur directe (raccourci)"""
        return self.auth.create_user(name, email, password)

    def get_user_info(self) -> Dict[str, Any]:
        """Méthode d'information utilisateur directe (raccourci)"""
        return self.auth.get_user_info()

    def is_authenticated(self) -> bool:
        """Vérifie si le client est authentifié"""
        return bool(self.token)

    def set_token(self, token: str) -> None:
        """Définit le token d'authentification"""
        self.token = token
        logger.info("Token défini pour le client API", token_length=len(token))

    def clear_token(self) -> None:
        """Supprime le token d'authentification"""
        self.token = None
        logger.info("Token supprimé du client API")

    # Méthodes de convenance pour un accès direct aux fonctionnalités principales
    def get_all_data(self) -> Dict[str, Any]:
        """Récupère toutes les données (utilisateurs et VMs) et les associe

        Returns:
            Dict contenant les utilisateurs avec leurs VMs associées

        Raises:
            VMsFetchError: Si la récupération des données échoue
        """
        logger.info("Récupération de toutes les données via API unifiée")
        users = self.users.get()
        vms = self.vms.get()
        self.users.add_vms_to_users(users, vms)
        return {
            "users": users,
            "vms": vms,
            "total_users": len(users),
            "total_vms": len(vms),
            "users_with_vms": len([u for u in users if u.get("vms")]),
        }

    def get_user_data(self, user_id: int) -> Dict[str, Any]:
        """Récupère un utilisateur avec ses VMs associées

        Args:
            user_id: ID de l'utilisateur

        Returns:
            Dict contenant l'utilisateur avec ses VMs

        Raises:
            UsersFetchError: Si l'utilisateur n'existe pas
        """
        logger.info(
            "Récupération des données d'un utilisateur via API unifiée", user_id=user_id
        )
        user = self.users.get_user(user_id)

        # Récupérer toutes les VMs et trouver celles de cet utilisateur
        all_vms = self.vms.get()
        user_vms = [vm for vm in all_vms if vm.get("user_id") == user_id]
        user["vms"] = user_vms

        return {"user": user, "vm_count": len(user_vms), "vms": user_vms}

    def get_vm_data(self, vm_id: int) -> Dict[str, Any]:
        """Récupère une VM avec les informations de son utilisateur

        Args:
            vm_id: ID de la VM

        Returns:
            Dict contenant la VM avec les infos utilisateur

        Raises:
            VMsFetchError: Si la VM n'existe pas
        """
        logger.info("Récupération des données d'une VM via API unifiée", vm_id=vm_id)
        vm = self.vms.get_vm(vm_id)

        # Récupérer les informations de l'utilisateur propriétaire
        if vm.get("user_id"):
            try:
                owner = self.users.get_user(vm["user_id"])
                vm["owner"] = owner
            except Exception:
                vm["owner"] = None

        return vm

    def __repr__(self) -> str:
        """Représentation du client API"""
        return f"ApiClient(base_url='{self.base_url}', authenticated={self.is_authenticated()})"

Client API unifié avec interface fluide

Initialise le client API

Args

base_url

URL de base de l'API (utilise config par défaut)

token

Token d'authentification (optionnel)

Methods

def clear_token(self) ‑> None

Expand source code

def clear_token(self) -> None:
    """Supprime le token d'authentification"""
    self.token = None
    logger.info("Token supprimé du client API")

Supprime le token d'authentification

def create_user(self, name: str, email: str, password: str) ‑> str

Expand source code

def create_user(self, name: str, email: str, password: str) -> str:
    """Méthode de création d'utilisateur directe (raccourci)"""
    return self.auth.create_user(name, email, password)

Méthode de création d'utilisateur directe (raccourci)

def get_all_data(self) ‑> Dict[str, Any]

Expand source code

def get_all_data(self) -> Dict[str, Any]:
    """Récupère toutes les données (utilisateurs et VMs) et les associe

    Returns:
        Dict contenant les utilisateurs avec leurs VMs associées

    Raises:
        VMsFetchError: Si la récupération des données échoue
    """
    logger.info("Récupération de toutes les données via API unifiée")
    users = self.users.get()
    vms = self.vms.get()
    self.users.add_vms_to_users(users, vms)
    return {
        "users": users,
        "vms": vms,
        "total_users": len(users),
        "total_vms": len(vms),
        "users_with_vms": len([u for u in users if u.get("vms")]),
    }

Récupère toutes les données (utilisateurs et VMs) et les associe

Returns

Dict contenant les utilisateurs avec leurs VMs associées

Raises

VMsFetchError

Si la récupération des données échoue

def get_user_data(self, user_id: int) ‑> Dict[str, Any]

Expand source code

def get_user_data(self, user_id: int) -> Dict[str, Any]:
    """Récupère un utilisateur avec ses VMs associées

    Args:
        user_id: ID de l'utilisateur

    Returns:
        Dict contenant l'utilisateur avec ses VMs

    Raises:
        UsersFetchError: Si l'utilisateur n'existe pas
    """
    logger.info(
        "Récupération des données d'un utilisateur via API unifiée", user_id=user_id
    )
    user = self.users.get_user(user_id)

    # Récupérer toutes les VMs et trouver celles de cet utilisateur
    all_vms = self.vms.get()
    user_vms = [vm for vm in all_vms if vm.get("user_id") == user_id]
    user["vms"] = user_vms

    return {"user": user, "vm_count": len(user_vms), "vms": user_vms}

Récupère un utilisateur avec ses VMs associées

Args

user_id

ID de l'utilisateur

Returns

Dict contenant l'utilisateur avec ses VMs

Raises

UsersFetchError

Si l'utilisateur n'existe pas

def get_user_info(self) ‑> Dict[str, Any]

Expand source code

def get_user_info(self) -> Dict[str, Any]:
    """Méthode d'information utilisateur directe (raccourci)"""
    return self.auth.get_user_info()

Méthode d'information utilisateur directe (raccourci)

def get_vm_data(self, vm_id: int) ‑> Dict[str, Any]

Expand source code

def get_vm_data(self, vm_id: int) -> Dict[str, Any]:
    """Récupère une VM avec les informations de son utilisateur

    Args:
        vm_id: ID de la VM

    Returns:
        Dict contenant la VM avec les infos utilisateur

    Raises:
        VMsFetchError: Si la VM n'existe pas
    """
    logger.info("Récupération des données d'une VM via API unifiée", vm_id=vm_id)
    vm = self.vms.get_vm(vm_id)

    # Récupérer les informations de l'utilisateur propriétaire
    if vm.get("user_id"):
        try:
            owner = self.users.get_user(vm["user_id"])
            vm["owner"] = owner
        except Exception:
            vm["owner"] = None

    return vm

Récupère une VM avec les informations de son utilisateur

Args

vm_id

ID de la VM

Returns

Dict contenant la VM avec les infos utilisateur

Raises

VMsFetchError

Si la VM n'existe pas

def is_authenticated(self) ‑> bool

Expand source code

def is_authenticated(self) -> bool:
    """Vérifie si le client est authentifié"""
    return bool(self.token)

Vérifie si le client est authentifié

def login(self, email: str, password: str) ‑> str

Expand source code

def login(self, email: str, password: str) -> str:
    """Méthode de connexion directe (raccourci)"""
    return self.auth.login(email, password)

Méthode de connexion directe (raccourci)

def set_token(self, token: str) ‑> None

Expand source code

def set_token(self, token: str) -> None:
    """Définit le token d'authentification"""
    self.token = token
    logger.info("Token défini pour le client API", token_length=len(token))

Définit le token d'authentification

class AuthAPI (api_client: ApiClient)

Expand source code

class AuthAPI:
    """Interface pour les opérations d'authentification"""

    def __init__(self, api_client: "ApiClient"):
        self._api = api_client

    def login(self, email: str, password: str) -> str:
        """Connexion d'un utilisateur

        Raises:
            UserLoginError: Si la connexion échoue
        """
        logger.info("Connexion utilisateur via API unifiée", email=email)
        token = self._api._auth.login_user(email, password)
        self._api.token = token
        logger.info("Connexion réussie via API unifiée", email=email)
        return token

    def create_user(self, name: str, email: str, password: str) -> str:
        """Création d'un utilisateur

        Raises:
            UserCreationError: Si la création d'utilisateur échoue
        """
        logger.info("Création utilisateur via API unifiée", email=email, name=name)
        token = self._api._auth.create_user(name, email, password)
        self._api.token = token
        logger.info("Utilisateur créé avec succès via API unifiée", email=email)
        return token

    def get_user_info(self) -> Dict[str, Any]:
        """Récupère les informations de l'utilisateur connecté

        Raises:
            TokenError: Si aucun token n'est disponible
            UserInfoError: Si la récupération des informations échoue
        """
        if not self._api.token:
            logger.warning(
                "Aucun token disponible pour récupérer les informations utilisateur"
            )
            raise TokenError(
                "Aucun token disponible pour récupérer les informations utilisateur"
            )

        logger.info("Récupération des informations utilisateur via API unifiée")
        return self._api._auth.get_logged_user_info(self._api.token)

Interface pour les opérations d'authentification

Methods

def create_user(self, name: str, email: str, password: str) ‑> str

Expand source code

def create_user(self, name: str, email: str, password: str) -> str:
    """Création d'un utilisateur

    Raises:
        UserCreationError: Si la création d'utilisateur échoue
    """
    logger.info("Création utilisateur via API unifiée", email=email, name=name)
    token = self._api._auth.create_user(name, email, password)
    self._api.token = token
    logger.info("Utilisateur créé avec succès via API unifiée", email=email)
    return token

Création d'un utilisateur

Raises

UserCreationError

Si la création d'utilisateur échoue

def get_user_info(self) ‑> Dict[str, Any]

Expand source code

def get_user_info(self) -> Dict[str, Any]:
    """Récupère les informations de l'utilisateur connecté

    Raises:
        TokenError: Si aucun token n'est disponible
        UserInfoError: Si la récupération des informations échoue
    """
    if not self._api.token:
        logger.warning(
            "Aucun token disponible pour récupérer les informations utilisateur"
        )
        raise TokenError(
            "Aucun token disponible pour récupérer les informations utilisateur"
        )

    logger.info("Récupération des informations utilisateur via API unifiée")
    return self._api._auth.get_logged_user_info(self._api.token)

Récupère les informations de l'utilisateur connecté

Raises

TokenError

Si aucun token n'est disponible

UserInfoError

Si la récupération des informations échoue

def login(self, email: str, password: str) ‑> str

Expand source code

def login(self, email: str, password: str) -> str:
    """Connexion d'un utilisateur

    Raises:
        UserLoginError: Si la connexion échoue
    """
    logger.info("Connexion utilisateur via API unifiée", email=email)
    token = self._api._auth.login_user(email, password)
    self._api.token = token
    logger.info("Connexion réussie via API unifiée", email=email)
    return token

Connexion d'un utilisateur

Raises

UserLoginError

Si la connexion échoue

class TokenError (message: str = 'Erreur liée au token',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
token_length: int | None = None)

Expand source code

class TokenError(DemoAPIException):
    """Exception levée lors d'erreurs liées aux tokens"""

    def __init__(
        self,
        message: str = "Erreur liée au token",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        token_length: Optional[int] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.token_length = token_length

Exception levée lors d'erreurs liées aux tokens

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class UserCreationError (message: str = "Erreur lors de la création d'utilisateur",
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
email: str | None = None)

Expand source code

class UserCreationError(DemoAPIException):
    """Exception levée lors d'erreurs de création d'utilisateur"""

    def __init__(
        self,
        message: str = "Erreur lors de la création d'utilisateur",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        email: Optional[str] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.email = email

Exception levée lors d'erreurs de création d'utilisateur

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class UserDeleteError (message: str = "Erreur lors de la suppression de l'utilisateur",
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
user_id: int | None = None)

Expand source code

class UserDeleteError(DemoAPIException):
    """Exception levée lors d'erreurs de suppression d'utilisateur"""

    def __init__(
        self,
        message: str = "Erreur lors de la suppression de l'utilisateur",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        user_id: Optional[int] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.user_id = user_id

Exception levée lors d'erreurs de suppression d'utilisateur

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class UserInfoError (message: str = 'Erreur lors de la récupération des informations utilisateur',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
token_length: int | None = None)

Expand source code

class UserInfoError(DemoAPIException):
    """Exception levée lors d'erreurs de récupération d'informations utilisateur"""

    def __init__(
        self,
        message: str = "Erreur lors de la récupération des informations utilisateur",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        token_length: Optional[int] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.token_length = token_length

Exception levée lors d'erreurs de récupération d'informations utilisateur

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class UserLoginError (message: str = 'Erreur lors de la connexion utilisateur',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
email: str | None = None)

Expand source code

class UserLoginError(DemoAPIException):
    """Exception levée lors d'erreurs de connexion utilisateur"""

    def __init__(
        self,
        message: str = "Erreur lors de la connexion utilisateur",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        email: Optional[str] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.email = email

Exception levée lors d'erreurs de connexion utilisateur

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class UserUpdateError (message: str = "Erreur lors de la mise à jour de l'utilisateur",
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
user_id: int | None = None)

Expand source code

class UserUpdateError(DemoAPIException):
    """Exception levée lors d'erreurs de mise à jour d'utilisateur"""

    def __init__(
        self,
        message: str = "Erreur lors de la mise à jour de l'utilisateur",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        user_id: Optional[int] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.user_id = user_id

Exception levée lors d'erreurs de mise à jour d'utilisateur

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class UsersAPI (api_client: ApiClient)

Expand source code

class UsersAPI:
    """Interface pour les opérations sur les utilisateurs"""

    def __init__(self, api_client: "ApiClient"):
        self._api = api_client

    def get(self) -> List[Dict[str, Any]]:
        """Récupère la liste des utilisateurs"""
        logger.info("Récupération des utilisateurs via API unifiée")
        return get_users(self._api.base_url)

    def add_vms_to_users(self, users: List[Dict], vms: List[Dict]) -> None:
        """Associe les VMs aux utilisateurs"""
        logger.info("Association des VMs aux utilisateurs via API unifiée")
        add_vms_to_users(users, vms)

    def get_user(self, user_id: int) -> Dict[str, Any]:
        """Récupère un utilisateur spécifique par son ID

        Args:
            user_id: ID de l'utilisateur

        Returns:
            Dict contenant les informations de l'utilisateur

        Raises:
            UsersFetchError: Si la récupération de l'utilisateur échoue
        """
        logger.info(
            "Récupération d'un utilisateur spécifique via API unifiée", user_id=user_id
        )
        return get_user(self._api.base_url, user_id)

    def create_user(
        self, name: str, email: str, password: Optional[str] = None
    ) -> Dict[str, Any]:
        """Crée un nouvel utilisateur

        Args:
            name: Nom de l'utilisateur
            email: Email de l'utilisateur
            password: Mot de passe optionnel

        Returns:
            Dict contenant les informations de l'utilisateur créé

        Raises:
            UserCreationError: Si la création de l'utilisateur échoue
        """
        logger.info("Création d'un utilisateur via API unifiée", name=name, email=email)
        return create_user(self._api.base_url, self._api.token, name, email, password)

    def update_user(self, user_id: int, updates: Dict[str, Any]) -> Dict[str, Any]:
        """Met à jour un utilisateur existant

        Args:
            user_id: ID de l'utilisateur
            updates: Données à mettre à jour

        Returns:
            Dict contenant les informations de l'utilisateur mis à jour

        Raises:
            UserUpdateError: Si la mise à jour échoue
        """
        logger.info("Mise à jour d'un utilisateur via API unifiée", user_id=user_id)
        return update_user(self._api.base_url, self._api.token, user_id, updates)

    def delete_user(self, user_id: int) -> Dict[str, Any]:
        """Supprime un utilisateur

        Args:
            user_id: ID de l'utilisateur

        Returns:
            Dict contenant le résultat de la suppression

        Raises:
            UserDeleteError: Si la suppression échoue
        """
        logger.info("Suppression d'un utilisateur via API unifiée", user_id=user_id)
        return delete_user(self._api.base_url, self._api.token, user_id)

    def create_vm(
        self,
        user_id: int,
        name: str,
        operating_system: str,
        cpu_cores: int,
        ram_gb: int,
        disk_gb: int,
        status: str = "stopped",
    ) -> Dict[str, Any]:
        """Crée une VM pour un utilisateur

        Raises:
            VMCreationError: Si la création de la VM échoue
        """
        logger.info(
            "Création de VM via API unifiée",
            user_id=user_id,
            name=name,
            operating_system=operating_system,
        )
        return create_vm(
            self._api.token,
            self._api.base_url,
            user_id=user_id,
            name=name,
            operating_system=operating_system,
            cpu_cores=cpu_cores,
            ram_gb=ram_gb,
            disk_gb=disk_gb,
            status=status,
        )

Interface pour les opérations sur les utilisateurs

Methods

def add_vms_to_users(self, users: List[Dict], vms: List[Dict]) ‑> None

Expand source code

def add_vms_to_users(self, users: List[Dict], vms: List[Dict]) -> None:
    """Associe les VMs aux utilisateurs"""
    logger.info("Association des VMs aux utilisateurs via API unifiée")
    add_vms_to_users(users, vms)

Associe les VMs aux utilisateurs

def create_user(self, name: str, email: str, password: str | None = None) ‑> Dict[str, Any]

Expand source code

def create_user(
    self, name: str, email: str, password: Optional[str] = None
) -> Dict[str, Any]:
    """Crée un nouvel utilisateur

    Args:
        name: Nom de l'utilisateur
        email: Email de l'utilisateur
        password: Mot de passe optionnel

    Returns:
        Dict contenant les informations de l'utilisateur créé

    Raises:
        UserCreationError: Si la création de l'utilisateur échoue
    """
    logger.info("Création d'un utilisateur via API unifiée", name=name, email=email)
    return create_user(self._api.base_url, self._api.token, name, email, password)

Crée un nouvel utilisateur

Args

name

Nom de l'utilisateur

email

Email de l'utilisateur

password

Mot de passe optionnel

Returns

Dict contenant les informations de l'utilisateur créé

Raises

UserCreationError

Si la création de l'utilisateur échoue

def create_vm(self,
user_id: int,
name: str,
operating_system: str,
cpu_cores: int,
ram_gb: int,
disk_gb: int,
status: str = 'stopped') ‑> Dict[str, Any]

Expand source code

def create_vm(
    self,
    user_id: int,
    name: str,
    operating_system: str,
    cpu_cores: int,
    ram_gb: int,
    disk_gb: int,
    status: str = "stopped",
) -> Dict[str, Any]:
    """Crée une VM pour un utilisateur

    Raises:
        VMCreationError: Si la création de la VM échoue
    """
    logger.info(
        "Création de VM via API unifiée",
        user_id=user_id,
        name=name,
        operating_system=operating_system,
    )
    return create_vm(
        self._api.token,
        self._api.base_url,
        user_id=user_id,
        name=name,
        operating_system=operating_system,
        cpu_cores=cpu_cores,
        ram_gb=ram_gb,
        disk_gb=disk_gb,
        status=status,
    )

Crée une VM pour un utilisateur

Raises

VMCreationError

Si la création de la VM échoue

def delete_user(self, user_id: int) ‑> Dict[str, Any]

Expand source code

def delete_user(self, user_id: int) -> Dict[str, Any]:
    """Supprime un utilisateur

    Args:
        user_id: ID de l'utilisateur

    Returns:
        Dict contenant le résultat de la suppression

    Raises:
        UserDeleteError: Si la suppression échoue
    """
    logger.info("Suppression d'un utilisateur via API unifiée", user_id=user_id)
    return delete_user(self._api.base_url, self._api.token, user_id)

Supprime un utilisateur

Args

user_id

ID de l'utilisateur

Returns

Dict contenant le résultat de la suppression

Raises

UserDeleteError

Si la suppression échoue

def get(self) ‑> List[Dict[str, Any]]

Expand source code

def get(self) -> List[Dict[str, Any]]:
    """Récupère la liste des utilisateurs"""
    logger.info("Récupération des utilisateurs via API unifiée")
    return get_users(self._api.base_url)

Récupère la liste des utilisateurs

def get_user(self, user_id: int) ‑> Dict[str, Any]

Expand source code

def get_user(self, user_id: int) -> Dict[str, Any]:
    """Récupère un utilisateur spécifique par son ID

    Args:
        user_id: ID de l'utilisateur

    Returns:
        Dict contenant les informations de l'utilisateur

    Raises:
        UsersFetchError: Si la récupération de l'utilisateur échoue
    """
    logger.info(
        "Récupération d'un utilisateur spécifique via API unifiée", user_id=user_id
    )
    return get_user(self._api.base_url, user_id)

Récupère un utilisateur spécifique par son ID

Args

user_id

ID de l'utilisateur

Returns

Dict contenant les informations de l'utilisateur

Raises

UsersFetchError

Si la récupération de l'utilisateur échoue

def update_user(self, user_id: int, updates: Dict[str, Any]) ‑> Dict[str, Any]

Expand source code

def update_user(self, user_id: int, updates: Dict[str, Any]) -> Dict[str, Any]:
    """Met à jour un utilisateur existant

    Args:
        user_id: ID de l'utilisateur
        updates: Données à mettre à jour

    Returns:
        Dict contenant les informations de l'utilisateur mis à jour

    Raises:
        UserUpdateError: Si la mise à jour échoue
    """
    logger.info("Mise à jour d'un utilisateur via API unifiée", user_id=user_id)
    return update_user(self._api.base_url, self._api.token, user_id, updates)

Met à jour un utilisateur existant

Args

user_id

ID de l'utilisateur

updates

Données à mettre à jour

Returns

Dict contenant les informations de l'utilisateur mis à jour

Raises

UserUpdateError

Si la mise à jour échoue

class UsersFetchError (message: str = 'Erreur lors de la récupération des utilisateurs',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
base_url: str | None = None)

Expand source code

class UsersFetchError(DemoAPIException):
    """Exception levée lors d'erreurs de récupération des utilisateurs"""

    def __init__(
        self,
        message: str = "Erreur lors de la récupération des utilisateurs",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        base_url: Optional[str] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.base_url = base_url

Exception levée lors d'erreurs de récupération des utilisateurs

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class VMCreationError (message: str = 'Erreur lors de la création de VM',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
user_id: int | None = None,
vm_name: str | None = None)

Expand source code

class VMCreationError(DemoAPIException):
    """Exception levée lors d'erreurs de création de VM"""

    def __init__(
        self,
        message: str = "Erreur lors de la création de VM",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        user_id: Optional[int] = None,
        vm_name: Optional[str] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.user_id = user_id
        self.vm_name = vm_name

Exception levée lors d'erreurs de création de VM

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class VMDeleteError (message: str = 'Erreur lors de la suppression de la VM',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
vm_id: int | None = None)

Expand source code

class VMDeleteError(DemoAPIException):
    """Exception levée lors d'erreurs de suppression de VM"""

    def __init__(
        self,
        message: str = "Erreur lors de la suppression de la VM",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        vm_id: Optional[int] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.vm_id = vm_id

Exception levée lors d'erreurs de suppression de VM

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class VMUpdateError (message: str = 'Erreur lors de la mise à jour de la VM',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
vm_id: int | None = None)

Expand source code

class VMUpdateError(DemoAPIException):
    """Exception levée lors d'erreurs de mise à jour de VM"""

    def __init__(
        self,
        message: str = "Erreur lors de la mise à jour de la VM",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        vm_id: Optional[int] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.vm_id = vm_id

Exception levée lors d'erreurs de mise à jour de VM

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException

class VMsAPI (api_client: ApiClient)

Expand source code

class VMsAPI:
    """Interface pour les opérations sur les VMs"""

    def __init__(self, api_client: "ApiClient"):
        self._api = api_client

    def get(self) -> List[Dict[str, Any]]:
        """Récupère la liste des VMs"""
        logger.info("Récupération des VMs via API unifiée")
        return get_vms(self._api.base_url)

    def create(
        self,
        user_id: int,
        name: str,
        operating_system: str,
        cpu_cores: int,
        ram_gb: int,
        disk_gb: int,
        status: str = "stopped",
    ) -> Dict[str, Any]:
        """Crée une VM

        Raises:
            VMCreationError: Si la création de la VM échoue
        """
        logger.info(
            "Création de VM via API unifiée",
            user_id=user_id,
            name=name,
            operating_system=operating_system,
        )
        return create_vm(
            self._api.token,
            self._api.base_url,
            user_id=user_id,
            name=name,
            operating_system=operating_system,
            cpu_cores=cpu_cores,
            ram_gb=ram_gb,
            disk_gb=disk_gb,
            status=status,
        )

    def get_vm(self, vm_id: int) -> Dict[str, Any]:
        """Récupère une VM spécifique par son ID

        Args:
            vm_id: ID de la VM

        Returns:
            Dict contenant les informations de la VM

        Raises:
            VMsFetchError: Si la récupération de la VM échoue
        """
        logger.info("Récupération d'une VM spécifique via API unifiée", vm_id=vm_id)
        return get_vm(self._api.base_url, vm_id)

    def update(self, vm_id: int, updates: Dict[str, Any]) -> Dict[str, Any]:
        """Met à jour une VM existante

        Args:
            vm_id: ID de la VM
            updates: Données à mettre à jour

        Returns:
            Dict contenant les informations de la VM mise à jour

        Raises:
            VMUpdateError: Si la mise à jour de la VM échoue
        """
        logger.info("Mise à jour d'une VM via API unifiée", vm_id=vm_id)
        return update_vm(self._api.base_url, self._api.token, vm_id, updates)

    def delete(self, vm_id: int) -> Dict[str, Any]:
        """Supprime une VM

        Args:
            vm_id: ID de la VM

        Returns:
            Dict contenant le résultat de la suppression

        Raises:
            VMDeleteError: Si la suppression échoue
        """
        logger.info("Suppression d'une VM via API unifiée", vm_id=vm_id)
        return delete_vm(self._api.base_url, self._api.token, vm_id)

    def attach_to_user(self, vm_id: int, user_id: int) -> Dict[str, Any]:
        """Associe une VM à un utilisateur

        Args:
            vm_id: ID de la VM
            user_id: ID de l'utilisateur

        Returns:
            Dict contenant le résultat de l'association

        Raises:
            VMUpdateError: Si l'association échoue
        """
        logger.info(
            "Association VM-utilisateur via API unifiée", vm_id=vm_id, user_id=user_id
        )
        return attach_vm_to_user(self._api.base_url, self._api.token, vm_id, user_id)

    def stop(self, vm_id: int) -> Dict[str, Any]:
        """Arrête une VM

        Args:
            vm_id: ID de la VM

        Returns:
            Dict contenant le résultat de l'arrêt

        Raises:
            VMUpdateError: Si l'arrêt de la VM échoue
        """
        logger.info("Arrêt d'une VM via API unifiée", vm_id=vm_id)
        return stop_vm(self._api.base_url, self._api.token, vm_id)

Interface pour les opérations sur les VMs

Methods

def attach_to_user(self, vm_id: int, user_id: int) ‑> Dict[str, Any]

Expand source code

def attach_to_user(self, vm_id: int, user_id: int) -> Dict[str, Any]:
    """Associe une VM à un utilisateur

    Args:
        vm_id: ID de la VM
        user_id: ID de l'utilisateur

    Returns:
        Dict contenant le résultat de l'association

    Raises:
        VMUpdateError: Si l'association échoue
    """
    logger.info(
        "Association VM-utilisateur via API unifiée", vm_id=vm_id, user_id=user_id
    )
    return attach_vm_to_user(self._api.base_url, self._api.token, vm_id, user_id)

Associe une VM à un utilisateur

Args

vm_id

ID de la VM

user_id

ID de l'utilisateur

Returns

Dict contenant le résultat de l'association

Raises

VMUpdateError

Si l'association échoue

def create(self,
user_id: int,
name: str,
operating_system: str,
cpu_cores: int,
ram_gb: int,
disk_gb: int,
status: str = 'stopped') ‑> Dict[str, Any]

Expand source code

def create(
    self,
    user_id: int,
    name: str,
    operating_system: str,
    cpu_cores: int,
    ram_gb: int,
    disk_gb: int,
    status: str = "stopped",
) -> Dict[str, Any]:
    """Crée une VM

    Raises:
        VMCreationError: Si la création de la VM échoue
    """
    logger.info(
        "Création de VM via API unifiée",
        user_id=user_id,
        name=name,
        operating_system=operating_system,
    )
    return create_vm(
        self._api.token,
        self._api.base_url,
        user_id=user_id,
        name=name,
        operating_system=operating_system,
        cpu_cores=cpu_cores,
        ram_gb=ram_gb,
        disk_gb=disk_gb,
        status=status,
    )

Crée une VM

Raises

VMCreationError

Si la création de la VM échoue

def delete(self, vm_id: int) ‑> Dict[str, Any]

Expand source code

def delete(self, vm_id: int) -> Dict[str, Any]:
    """Supprime une VM

    Args:
        vm_id: ID de la VM

    Returns:
        Dict contenant le résultat de la suppression

    Raises:
        VMDeleteError: Si la suppression échoue
    """
    logger.info("Suppression d'une VM via API unifiée", vm_id=vm_id)
    return delete_vm(self._api.base_url, self._api.token, vm_id)

Supprime une VM

Args

vm_id

ID de la VM

Returns

Dict contenant le résultat de la suppression

Raises

VMDeleteError

Si la suppression échoue

def get(self) ‑> List[Dict[str, Any]]

Expand source code

def get(self) -> List[Dict[str, Any]]:
    """Récupère la liste des VMs"""
    logger.info("Récupération des VMs via API unifiée")
    return get_vms(self._api.base_url)

Récupère la liste des VMs

def get_vm(self, vm_id: int) ‑> Dict[str, Any]

Expand source code

def get_vm(self, vm_id: int) -> Dict[str, Any]:
    """Récupère une VM spécifique par son ID

    Args:
        vm_id: ID de la VM

    Returns:
        Dict contenant les informations de la VM

    Raises:
        VMsFetchError: Si la récupération de la VM échoue
    """
    logger.info("Récupération d'une VM spécifique via API unifiée", vm_id=vm_id)
    return get_vm(self._api.base_url, vm_id)

Récupère une VM spécifique par son ID

Args

vm_id

ID de la VM

Returns

Dict contenant les informations de la VM

Raises

VMsFetchError

Si la récupération de la VM échoue

def stop(self, vm_id: int) ‑> Dict[str, Any]

Expand source code

def stop(self, vm_id: int) -> Dict[str, Any]:
    """Arrête une VM

    Args:
        vm_id: ID de la VM

    Returns:
        Dict contenant le résultat de l'arrêt

    Raises:
        VMUpdateError: Si l'arrêt de la VM échoue
    """
    logger.info("Arrêt d'une VM via API unifiée", vm_id=vm_id)
    return stop_vm(self._api.base_url, self._api.token, vm_id)

Arrête une VM

Args

vm_id

ID de la VM

Returns

Dict contenant le résultat de l'arrêt

Raises

VMUpdateError

Si l'arrêt de la VM échoue

def update(self, vm_id: int, updates: Dict[str, Any]) ‑> Dict[str, Any]

Expand source code

def update(self, vm_id: int, updates: Dict[str, Any]) -> Dict[str, Any]:
    """Met à jour une VM existante

    Args:
        vm_id: ID de la VM
        updates: Données à mettre à jour

    Returns:
        Dict contenant les informations de la VM mise à jour

    Raises:
        VMUpdateError: Si la mise à jour de la VM échoue
    """
    logger.info("Mise à jour d'une VM via API unifiée", vm_id=vm_id)
    return update_vm(self._api.base_url, self._api.token, vm_id, updates)

Met à jour une VM existante

Args

vm_id

ID de la VM

updates

Données à mettre à jour

Returns

Dict contenant les informations de la VM mise à jour

Raises

VMUpdateError

Si la mise à jour de la VM échoue

class VMsFetchError (message: str = 'Erreur lors de la récupération des VMs',
status_code: int | None = None,
response_data: Dict[str, Any] | None = None,
base_url: str | None = None)

Expand source code

class VMsFetchError(DemoAPIException):
    """Exception levée lors d'erreurs de récupération des VMs"""

    def __init__(
        self,
        message: str = "Erreur lors de la récupération des VMs",
        status_code: Optional[int] = None,
        response_data: Optional[Dict[str, Any]] = None,
        base_url: Optional[str] = None,
    ):
        super().__init__(message, status_code, response_data)
        self.base_url = base_url

Exception levée lors d'erreurs de récupération des VMs

Ancestors

• DemoAPIException
• builtins.Exception
• builtins.BaseException