from __future__ import annotations
from enum import Enum
from typing import Any
from abc import abstractmethod, ABC
from ..utils import Request, Api, POST, GET, PUT, DELETE
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..utils import CloudContext
from .cloud_user import User
[docs]
class TokenType(Enum):
"""Enum defining different token types used for authentication.
Attributes:
Login (int): Login token type used for general authentication.
Job (int): Job token type used for job-specific operations.
API_Key (int): API key token type used for worker access.
"""
[docs]
class ACloudObject(ABC, dict):
"""Abstract base class for all cloud objects.
Provides common functionality for API interactions including CRUD operations
and endpoint management. All cloud objects (User, Job, Platform, etc.) should
inherit from this class.
"""
@classmethod
@abstractmethod
[docs]
def get_object_name(cls) -> str:
"""Return the name of the object type.
This method should be implemented by all subclasses to define
the object name used in API paths and object identification.
:return: Name of the object used in API paths
"""
raise NotImplementedError()
@classmethod
@abstractmethod
[docs]
def get_endpoint(cls) -> str:
"""Return the base API endpoint for this object type.
This method should be implemented by all subclasses to define
the API endpoint path for the object.
:return: API endpoint path for this object type
"""
raise NotImplementedError()
@classmethod
[docs]
def get_object_id_key(cls) -> str:
"""Return the key name used for the object's ID.
By default, creates the key by appending '_id' to the object name.
:return: Key name for the object's ID
"""
return cls.get_object_name()+"_id"
@classmethod
[docs]
def get_id_endpoint(cls) -> str:
"""Return the API endpoint for accessing a specific object instance.
Constructs an endpoint with the object ID parameter in the path.
:return: API endpoint with object ID parameter
"""
return cls.get_endpoint() + "/{" + cls.get_object_id_key()+"}"
@classmethod
[docs]
def get_id_endpoint_with_extension(cls, extension_path: str) -> str:
"""Return the API endpoint with an extension for a specific object instance.
Constructs an endpoint with the object ID parameter and additional path extension.
:param extension_path: Additional path to append after the object ID
:return: Extended API endpoint with object ID parameter
"""
return cls.get_endpoint() + "/{" + cls.get_object_id_key()+"}" + "/" + extension_path
[docs]
def get_object_id(self) -> str | int:
"""Return the ID of this object instance.
:return: The object's ID value
"""
return self.__dict__[self.get_object_id_key()]
@classmethod
@abstractmethod
[docs]
def get_api(cls, context: CloudContext) -> Api:
"""Return the API instance for this object type.
This method should be implemented by all subclasses to define
which API context to use for requests.
:param context: Cloud context containing API configurations
:return: API instance for making requests
"""
raise NotImplementedError()
@classmethod
[docs]
def get_token_method(cls) -> str:
"""Return the token type used for authentication with this object.
By default, uses the Login token type.
:return: Token type for authentication
"""
return TokenType.Login
[docs]
def keys(self) -> list[str]:
"""Return a list of all keys in this object's dictionary.
:return: List of dictionary keys
"""
return list(self.__dict__.keys())
[docs]
def update_from_dict(self, object_dict: dict[str]) -> None:
"""Update this object's attributes from a dictionary.
Special handling for 'id' key to map it to the appropriate object ID key.
:param object_dict: Dictionary with attributes to update
"""
for key in object_dict:
if key == "id":
self.__dict__[self.get_object_id_key()] = object_dict[key]
else:
self.__dict__[key] = object_dict[key]
[docs]
def get_create_request(self, context: CloudContext, user) -> Request:
"""Create a new request for creating this object.
Configures a POST request to the object's endpoint with appropriate
authentication and content type headers.
:param context: Cloud context for API configuration
:param user: User providing authentication
:return: Configured request object
"""
request = Request(api=self.get_api(context))
request.set_authorization_header(user.get_token(self.get_token_method()))
request.set_json_content_type()
request.set_method(POST)
request.set_url(self.get_endpoint())
return request
def _get_endpoint_id_request(self, context: CloudContext, user: User) -> Request:
"""Create a base request for operating on a specific object instance.
Helper method for creating requests that target a specific object ID.
:param context: Cloud context for API configuration
:param user: User providing authentication
:return: Configured request object with object ID in URL
"""
request = Request(api=self.get_api(context))
request.set_authorization_header(user.get_token(self.get_token_method()))
request.set_json_content_type()
request.set_url(self.get_id_endpoint(), {self.get_object_id_key(): self.get_object_id()})
return request
[docs]
def get_endpoint_id_request_with_extension(
self,
context: CloudContext,
user,
extension_path: str) -> Request:
"""Create a request for an extended endpoint for this object instance.
Creates a request that targets a specific object ID with an additional path extension.
:param context: Cloud context for API configuration
:param user: User providing authentication
:param extension_path: Additional path to append after the object ID
:return: Configured request object with extended URL
"""
request = self._get_endpoint_id_request(context, user)
request.set_url(self.get_id_endpoint_with_extension(extension_path),
{self.get_object_id_key(): self.get_object_id()})
return request
[docs]
def get_read_request(self, context: CloudContext, user: User) -> Request:
"""Create a request for reading this object instance.
Configures a GET request to access the object's data.
:param context: Cloud context for API configuration
:param user: User providing authentication
:return: Configured GET request
"""
request = self._get_endpoint_id_request(context, user)
request.set_method(GET)
return request
[docs]
def get_update_request(self, context: CloudContext, user: User) -> Request:
"""Create a request for updating this object instance.
Configures a PUT request to update the object's data.
:param context: Cloud context for API configuration
:param user: User providing authentication
:return: Configured PUT request
"""
request = self._get_endpoint_id_request(context, user)
request.set_method(PUT)
return request
[docs]
def get_delete_request(self, context: CloudContext, user: User) -> Request:
"""Create a request for deleting this object instance.
Configures a DELETE request to remove the object.
:param context: Cloud context for API configuration
:param user: User providing authentication
:return: Configured DELETE request
"""
request = self._get_endpoint_id_request(context, user)
request.set_method(DELETE)
return request
@classmethod
[docs]
def get_list_all_request(cls, context: CloudContext, user: User) -> Request:
"""Create a request for listing all objects of this type.
Configures a GET request to the base endpoint to retrieve all objects.
:param context: Cloud context for API configuration
:param user: User providing authentication
:return: Configured GET request for listing objects
"""
request = Request(api=cls.get_api(context))
request.set_authorization_header(user.get_token(TokenType.Login))
request.set_json_content_type()
request.set_method(GET)
request.set_url(cls.get_endpoint())
return request
def __getitem__(self, key: str) -> Any:
"""Dictionary-style access to object attributes.
:param key: Attribute name to access
:return: Value of the requested attribute
"""
return getattr(self, key)
def __setitem__(self, key, value) -> None:
"""Dictionary-style setting of object attributes.
:param key: Attribute name to set
:param value: Value to assign to the attribute
"""
self.__dict__[key] = value