# pylint: disable=W0718,C0302
"""Smarter Api ApiConnection Manifest handler"""
import logging
from datetime import datetime
from typing import TYPE_CHECKING, Optional, Type
from django.core.exceptions import MultipleObjectsReturned
from smarter.apps.account.models import Secret, UserProfile
from smarter.apps.account.utils import (
get_cached_admin_user_for_account,
smarter_cached_objects,
)
from smarter.apps.plugin.manifest.enum import SAMApiConnectionSpecConnectionKeys
from smarter.apps.plugin.manifest.models.api_connection.const import MANIFEST_KIND
from smarter.apps.plugin.manifest.models.api_connection.enum import AuthMethods
from smarter.apps.plugin.manifest.models.api_connection.model import SAMApiConnection
from smarter.apps.plugin.manifest.models.api_connection.spec import (
ApiConnection as PydanticApiConnection,
)
from smarter.apps.plugin.manifest.models.api_connection.spec import (
SAMApiConnectionSpec,
)
from smarter.apps.plugin.manifest.models.common.connection.metadata import (
SAMConnectionCommonMetadata,
)
from smarter.apps.plugin.manifest.models.common.connection.status import (
SAMConnectionCommonStatus,
)
from smarter.apps.plugin.models import ApiConnection
from smarter.apps.plugin.serializers import ApiConnectionSerializer
from smarter.common.exceptions import SmarterValueError
from smarter.common.utils import camel_to_snake
from smarter.lib import json
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.journal.enum import SmarterJournalCliCommands
from smarter.lib.journal.http import SmarterJournaledJsonResponse
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
from smarter.lib.manifest.broker import (
SAMBrokerErrorNotImplemented,
SAMBrokerErrorNotReady,
)
from smarter.lib.manifest.enum import (
SAMKeys,
SAMMetadataKeys,
SCLIResponseGet,
SCLIResponseGetData,
)
from . import SAMConnectionBrokerError
from .connection_base import SAMConnectionBaseBroker
if TYPE_CHECKING:
from django.http import HttpRequest
# pylint: disable=W0613
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return waffle.switch_is_active(SmarterWaffleSwitches.PLUGIN_LOGGING) or waffle.switch_is_active(
SmarterWaffleSwitches.MANIFEST_LOGGING
)
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
[docs]
class SAMApiConnectionBroker(SAMConnectionBaseBroker):
"""
Smarter API ApiConnection Manifest Broker.
This class is responsible for loading, validating, and parsing Smarter API YAML ApiConnection manifests, and initializing the corresponding Pydantic model. It provides generic services for ApiConnection objects, including instantiation, creation, update, and deletion.
:param loader: Manifest loader providing manifest data.
:type loader: Optional[ManifestLoader]
:param account: The account context for the connection.
:type account: Account
:param user_profile: The user profile associated with the connection.
:type user_profile: UserProfile
.. seealso::
:class:`smarter.apps.plugin.manifest.models.api_connection.model.SAMApiConnection`
:class:`smarter.apps.plugin.models.ApiConnection`
:class:`smarter.apps.plugin.serializers.ApiConnectionSerializer`
:class:`smarter.apps.plugin.manifest.brokers.SAMConnectionBrokerError`
**Example usage**::
broker = SAMApiConnectionBroker(loader=my_loader, account=my_account, user_profile=my_profile)
manifest = broker.manifest
orm_data = broker.manifest_to_django_orm()
"""
[docs]
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
if not self.ready:
if not self.loader and not self.manifest and not self.connection:
logger.warning(
"%s.__init__() No loader nor existing Connection provided for %s broker. Cannot initialize.",
self.formatted_class_name,
self.kind,
)
return
if self.loader and self.loader.manifest_kind != self.kind:
raise SAMBrokerErrorNotReady(
f"Loader manifest kind {self.loader.manifest_kind} does not match broker kind {self.kind}",
thing=self.kind,
)
if self.loader:
self._manifest = SAMApiConnection(
apiVersion=self.loader.manifest_api_version,
kind=self.loader.manifest_kind,
metadata=SAMConnectionCommonMetadata(**self.loader.manifest_metadata),
spec=SAMApiConnectionSpec(**self.loader.manifest_spec),
status=(
SAMConnectionCommonStatus(**self.loader.manifest_status)
if self.loader and self.loader.manifest_status
else None
),
)
if self._manifest:
logger.info(
"%s.__init__() initialized manifest from loader for %s %s",
self.formatted_class_name,
self.kind,
self.manifest.metadata.name,
)
msg = f"{self.formatted_class_name}.__init__() broker for {self.kind} {self.name} is {self.ready_state}."
if self.ready:
logger.info(msg)
else:
logger.warning(msg)
# override the base abstract manifest model with the ApiConnection model
_manifest: Optional[SAMApiConnection] = None
_pydantic_model: Type[SAMApiConnection] = SAMApiConnection
_connection: Optional[ApiConnection] = None
_api_key_secret: Optional[Secret] = None
_proxy_password_secret: Optional[Secret] = None
[docs]
def connection_init(self) -> None:
"""
Initialize the connection-related properties of the broker.
This method resets the internal state of the broker related to the ApiConnection instance and its associated secrets. It is useful when reloading or refreshing the connection data to ensure that stale references are cleared.
:return: None
:rtype: None
.. seealso::
:meth:`SAMApiConnectionBroker.connection`
:meth:`SAMApiConnectionBroker.api_key_secret`
:meth:`SAMApiConnectionBroker.proxy_password_secret`
**Example usage**::
broker.connection_init()
connection = broker.connection
"""
super().connection_init()
self._manifest = None
self._connection = None
self._api_key_secret = None
self._proxy_password_secret = None
###########################################################################
# Smarter abstract property implementations
###########################################################################
@property
def SerializerClass(self) -> Type[ApiConnectionSerializer]:
"""
Return the SerializerClass class for the broker.
This property provides the SerializerClass used to convert ApiConnection model instances to and from native Python datatypes, enabling validation and serialization for API responses and internal processing.
:return: The SerializerClass class for ApiConnection objects.
:rtype: Type[ApiConnectionSerializer]
.. seealso::
:class:`ApiConnectionSerializer`
:class:`ApiConnection`
:meth:`SAMApiConnectionBroker.manifest`
:meth:`SAMApiConnectionBroker.manifest_to_django_orm`
**Example usage**::
serializer_cls = broker.SerializerClass
SerializerClass = serializer_cls(api_connection_instance)
data = SerializerClass.data
"""
return ApiConnectionSerializer
@property
def formatted_class_name(self) -> str:
"""
Returns the formatted class name for logging purposes.
This property generates a human-readable class name string for use in log messages, making it easier to identify the source of log entries. It appends the specific broker class to the parent class name for clarity.
:return: Formatted class name string for logging.
:rtype: str
.. important::
Use this property in log statements to improve traceability and debugging.
.. seealso::
:meth:`SAMApiConnectionBroker.SerializerClass`
:meth:`SAMApiConnectionBroker.manifest`
:meth:`SAMApiConnectionBroker.ORMModelClass`
**Example usage**::
logger.info("%s: operation started", broker.formatted_class_name)
"""
return f"{__name__}.{SAMApiConnectionBroker.__name__}[{id(self)}]"
@property
def ORMMetaModelClass(self) -> Type[ApiConnection]:
"""
Return the Django ORM meta model class for the broker.
:return: The Django ORM meta model class definition for the broker.
:rtype: Type[ApiConnection]
"""
return ApiConnection
@property
def ORMModelClass(self) -> Type[ApiConnection]:
"""
Return the Django ORM model class for ApiConnection.
This property provides the class object used for persistent storage and manipulation of API connection data in the database. It is useful for type checking, introspection, and for creating or querying ApiConnection instances.
:return: The Django ORM model class for API connections.
:rtype: Type[smarter.apps.plugin.models.ApiConnection]
.. seealso::
:class:`smarter.apps.plugin.manifest.models.api_connection.spec.ApiConnection`
:meth:`SAMApiConnectionBroker.SerializerClass`
:meth:`SAMApiConnectionBroker.manifest`
**Example usage**::
model_cls = broker.ORMModelClass
all_connections = model_cls.objects.all()
"""
return ApiConnection
@property
def SAMModelClass(self) -> Type[SAMApiConnection]:
"""
Return the Pydantic model class for the broker.
:return: The Pydantic model class definition for the broker.
:rtype: Type[SAMApiConnection]
"""
return SAMApiConnection
@property
def kind(self) -> str:
return MANIFEST_KIND
@property
def manifest(self) -> Optional[SAMApiConnection]:
"""
Returns the manifest as a Pydantic model representing the Smarter API ApiConnection manifest.
This property initializes and returns a ``SAMApiConnection`` Pydantic model using data
loaded from the manifest loader. The manifest loader provides the manifest's API version,
kind, metadata, spec, and status, which are passed to the model constructor.
The top-level manifest model must be explicitly initialized, while child models
(such as metadata, spec, and status) are automatically cascade-initialized by Pydantic,
passing the relevant data to each child's constructor.
If the loader's manifest kind does not match the expected kind, a warning is logged
and the manifest is not initialized.
:return: The manifest as a ``SAMApiConnection`` Pydantic model, or ``None`` if not initialized.
:rtype: Optional[SAMApiConnection]
"""
if self._manifest:
if not isinstance(self._manifest, SAMApiConnection):
raise SAMConnectionBrokerError(
message=f"Invalid manifest type for {self.kind} broker: {type(self._manifest)}",
thing=self.kind,
)
return self._manifest
# 1.) prioritize manifest loader data if available. if it was provided
# in the request body then this is the authoritative source.
if self.loader and self.loader.manifest_kind == self.kind:
self._manifest = SAMApiConnection(
apiVersion=self.loader.manifest_api_version,
kind=self.loader.manifest_kind,
metadata=SAMConnectionCommonMetadata(**self.loader.manifest_metadata),
spec=SAMApiConnectionSpec(**self.loader.manifest_spec),
status=(
SAMConnectionCommonStatus(**self.loader.manifest_status)
if self.loader and self.loader.manifest_status
else None
),
)
logger.info("%s.manifest() initialized manifest from loader", self.formatted_class_name)
# 2.) next, (and only if a loader is not available) try to initialize
# from existing Account model if available
elif self.connection:
metadata = self.sam_connection_metadata()
if not metadata:
raise SAMBrokerErrorNotReady(
f"Metadata could not be built for account {self.account}. Cannot build manifest.",
thing=self.kind,
)
connection = PydanticApiConnection(
baseUrl=self.connection.base_url,
apiKey=self.connection.api_key.get_secret() if self.connection.api_key else None,
authMethod=self.connection.auth_method,
timeout=self.connection.timeout,
proxyProtocol=self.connection.proxy_protocol,
proxyHost=self.connection.proxy_host,
proxyPort=self.connection.proxy_port,
proxyUsername=self.connection.proxy_username,
proxyPassword=self.connection.proxy_password,
)
spec = SAMApiConnectionSpec(
connection=connection,
)
admin = get_cached_admin_user_for_account(account=self.account) # type: ignore
if not admin:
raise SAMBrokerErrorNotReady(
f"Admin user not found for account {self.account}. Cannot build manifest.",
thing=self.kind,
)
status = self.sam_connection_status()
self._manifest = SAMApiConnection(
apiVersion=self.api_version,
kind=self.kind,
metadata=metadata,
spec=spec,
status=status,
)
return self._manifest
else:
logger.warning(
"%s.manifest() could not initialize manifest. Expected %s but got %s",
self.formatted_class_name,
self.kind,
self.loader.manifest_kind if self.loader else None,
)
if not self._manifest:
logger.warning("%s.manifest could not be initialized", self.formatted_class_name)
return self._manifest
[docs]
def manifest_to_django_orm(self) -> dict:
"""
Transform the Smarter API User manifest into a Django ORM model.
This method converts the validated manifest data into a dictionary suitable for creating or updating a Django ORM `ApiConnection` instance. It handles field mapping, type conversion, and secret resolution for sensitive fields such as API keys and proxy passwords.
:returns: Dictionary of ORM-compatible fields for an `ApiConnection` model.
:rtype: dict
.. note::
- The returned dictionary includes all required fields for ORM persistence, with secrets resolved to their database IDs.
- The method automatically converts camelCase keys to snake_case for Django compatibility.
:raises SAMConnectionBrokerError:
If the manifest or its spec is missing or malformed
.. seealso::
:class:`ApiConnection`
:meth:`SAMApiConnectionBroker.manifest`
:meth:`SAMApiConnectionBroker.SerializerClass`
**Example usage**::
orm_data = broker.manifest_to_django_orm()
connection = ApiConnection(**orm_data)
connection.save()
"""
metadata = super().manifest_to_django_orm()
config_dump = self.manifest.spec.connection.model_dump() if self.manifest and self.manifest.spec else None
if not isinstance(config_dump, dict):
raise SAMConnectionBrokerError(
f"Manifest spec.connection is not a dict: {type(config_dump)}",
thing=self.kind,
)
config_dump = self.camel_to_snake(config_dump)
if not isinstance(config_dump, dict):
config_dump = json.loads(json.dumps(config_dump))
config_dump[SAMMetadataKeys.NAME.value] = (
self.manifest.metadata.name if self.manifest and self.manifest.metadata else None
)
config_dump[SAMMetadataKeys.DESCRIPTION.value] = (
self.manifest.metadata.description if self.manifest and self.manifest.metadata else None
)
config_dump[SAMMetadataKeys.VERSION.value] = (
self.manifest.metadata.version if self.manifest and self.manifest.metadata else None
)
config_dump[SAMKeys.KIND.value] = self.kind
if not self.user_profile:
raise SAMConnectionBrokerError(
"User profile is not set. Cannot retrieve or create secrets.",
thing=self.kind,
)
# retrieve the apiKey Secret
api_key_name = str(camel_to_snake(SAMApiConnectionSpecConnectionKeys.API_KEY.value))
if api_key_name:
try:
secret = Secret.objects.get(name=api_key_name, user_profile=self.user_profile)
config_dump[SAMApiConnectionSpecConnectionKeys.API_KEY.value] = secret.id if secret else None # type: ignore[assignment]
except Secret.DoesNotExist:
logger.warning(
"%s.manifest_to_django_orm() api key Secret %s not found for user %s",
self.formatted_class_name,
api_key_name,
self.user_profile.cached_user.username,
)
# retrieve the proxyUsername Secret, if it exists
proxy_password_name = str(camel_to_snake(SAMApiConnectionSpecConnectionKeys.PROXY_PASSWORD.value))
if proxy_password_name:
try:
secret = Secret.objects.get(name=proxy_password_name, user_profile=self.user_profile)
config_dump[SAMApiConnectionSpecConnectionKeys.PROXY_PASSWORD.value] = secret.id if secret else None # type: ignore[assignment]
except Secret.DoesNotExist:
logger.warning(
"%s.manifest_to_django_orm() proxy password Secret %s not found for user %s",
self.formatted_class_name,
proxy_password_name,
self.user_profile.cached_user.username,
)
return {**metadata, **config_dump}
@property
def api_key_secret(self) -> Optional[Secret]:
"""
Return the api_key secret for the ApiConnection.
This property retrieves the Django ORM `Secret` instance associated with the API key for the current connection. It resolves the secret either from the manifest or from the existing database record, depending on initialization context.
:return: The `Secret` object representing the API key, or `None` if not found.
:rtype: Optional[Secret]
.. attention::
If the secret cannot be found, a warning is logged and `None` is returned.
.. seealso::
:class:`Secret`
:meth:`SAMApiConnectionBroker.manifest`
:meth:`SAMApiConnectionBroker.connection`
**Example usage**::
api_key_secret = broker.api_key_secret
if api_key_secret:
print(api_key_secret.value)
"""
if self._api_key_secret:
return self._api_key_secret
try:
name = (
self.manifest.spec.connection.apiKey
if self.manifest and self.manifest.spec
else self.connection.api_key.name if self.connection and self.connection.api_key else None
)
self._api_key_secret = Secret.objects.get(name=name, user_profile=self.user_profile)
return self._api_key_secret
except Secret.DoesNotExist:
logger.warning(
"%s api_key Secret %s not found for account %s",
self.formatted_class_name,
name or "(name is missing)",
self.account,
)
return None
@property
def proxy_password_secret(self) -> Optional[Secret]:
"""
Return the proxy password secret for the ApiConnection.
This property retrieves the Django ORM `Secret` instance associated with the proxy password for the current connection. It resolves the secret either from the manifest or from the existing database record, depending on initialization context.
:return: The `Secret` object representing the proxy password, or `None` if not found.
:rtype: Optional[Secret]
.. attention::
- If the secret cannot be found, a warning is logged and `None` is returned.
.. seealso::
:class:`Secret`
:meth:`SAMApiConnectionBroker.manifest`
:meth:`SAMApiConnectionBroker.connection`
**Example usage**::
proxy_secret = broker.proxy_password_secret
if proxy_secret:
print(proxy_secret.value)
"""
if self._proxy_password_secret:
return self._proxy_password_secret
try:
name = (
self.manifest.spec.connection.proxyPassword
if self.manifest and self.manifest.spec
else (
self.connection.proxy_password.name if self.connection and self.connection.proxy_password else None
)
)
self._proxy_password_secret = Secret.objects.get(name=name, user_profile=self.user_profile)
return self._proxy_password_secret
except Secret.DoesNotExist:
logger.warning(
"%s proxy password Secret %s not found for account %s",
self.formatted_class_name,
name or "(name is missing)",
self.account,
)
return None
@property
def connection(self) -> Optional[ApiConnection]:
"""
Return the Django ORM `ApiConnection` instance for this broker.
This property retrieves the current `ApiConnection` object from the database using the account and name. If the connection does not exist, it attempts to create one from the manifest data. The returned object represents the persistent state of the API connection.
:return: The `ApiConnection` ORM instance, or `None` if not found or not created.
:rtype: Optional[smarter.apps.plugin.models.ApiConnection]
.. attention::
- If the connection cannot be found or created, an error is logged and `None` is returned.
.. seealso::
:class:`ApiConnection`
:meth:`SAMApiConnectionBroker.manifest`
:meth:`SAMApiConnectionBroker.manifest_to_django_orm`
:meth:`SAMApiConnectionBroker.api_key_secret`
:meth:`SAMApiConnectionBroker.proxy_password_secret`
**Example usage**::
connection = broker.connection
if connection:
print(connection.base_url)
connection.timeout = 60
connection.save()
"""
if self._connection:
return self._connection
name = str(self.camel_to_snake(self.name)) # type: ignore
if not name:
raise SmarterValueError(
f"Connection name is required to retrieve or create ApiConnection for {self.user_profile}"
)
try:
logger.debug(
"%s.connection() attempting ApiConnection with account %s and name %s",
self.formatted_class_name,
self.account,
name,
)
self._connection = ApiConnection.objects.get(user_profile=self.user_profile, name=name)
except MultipleObjectsReturned:
logger.debug(
"%s.connection() multiple ApiConnection objects found for %s and name %s",
self.formatted_class_name,
self.account,
name,
)
try:
self._connection = ApiConnection.objects.get(user_profile=self.user_profile, name=name)
except ApiConnection.DoesNotExist:
logger.debug(
"%s.connection() no ApiConnection found for user profile %s and name %s",
self.formatted_class_name,
self.user_profile,
name,
)
except ApiConnection.DoesNotExist:
try:
if self.user_profile:
admin_user = UserProfile.admin_for_account(self.user_profile.account)
admin_user_profile = UserProfile.get_cached_object(user=admin_user) # type: ignore
self._connection = ApiConnection.objects.get(user_profile=admin_user_profile, name=name)
except ApiConnection.DoesNotExist:
try:
self._connection = ApiConnection.objects.get(
user_profile=smarter_cached_objects.smarter_admin_user_profile, name=name
)
except ApiConnection.DoesNotExist:
logger.debug(
"%s.connection() no ApiConnection found for account %s and name %s",
self.formatted_class_name,
self.account,
name,
)
if not self._connection:
if self._manifest:
model_dump = (
self._manifest.spec.connection.model_dump() if self._manifest and self._manifest.spec else None
)
model_dump = self.camel_to_snake(model_dump) if isinstance(model_dump, dict) else model_dump
if not isinstance(model_dump, dict):
raise SAMConnectionBrokerError(
f"Manifest spec.connection is not a dict: {type(model_dump)}",
thing=self.kind,
)
# model_dump[SAMMetadataKeys.ACCOUNT.value] = self.account
model_dump[SAMMetadataKeys.NAME.value] = (
self.manifest.metadata.name if self.manifest and self.manifest.metadata else None
)
model_dump[SAMMetadataKeys.VERSION.value] = (
self.manifest.metadata.version if self.manifest and self.manifest.metadata else None
)
model_dump[SAMMetadataKeys.DESCRIPTION.value] = (
self.manifest.metadata.description if self.manifest and self.manifest.metadata else None
)
model_dump[SAMKeys.KIND.value] = self.kind
model_dump["api_key"] = self.api_key_secret
model_dump["user_profile"] = self.user_profile
self._connection = ApiConnection(**model_dump)
self._connection.save()
self._created = True
logger.info(
"%s.connection() created ApiConnection %s for account %s",
self.formatted_class_name,
self.name or "(name is missing)",
self.account or "(account is missing)",
)
else:
logger.error(
"%s.connection() ApiConnection %s not found for account %s",
self.formatted_class_name,
self.name or "(name is missing)",
self.account or "(account is missing)",
)
return self._connection
[docs]
def example_manifest(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Return an example ApiConnection manifest.
This method generates and returns a sample manifest for an ApiConnection, including all required fields and example values for authentication, connection, and metadata. The manifest is validated using the Pydantic model and returned as a JSON response.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: JSON response containing the example manifest.
:rtype: SmarterJournaledJsonResponse
.. seealso::
:class:`SAMApiConnection`
:class:`ApiConnection`
:class:`ApiConnectionSerializer`
:class:`AuthMethods`
:class:`SAMKeys`
:class:`SAMMetadataKeys`
:class:`SAMApiConnectionSpecConnectionKeys`
:class:`SmarterJournalCliCommands`
**Example usage**::
response = broker.example_manifest(request)
print(response.data)
"""
logger.debug(
"%s.example_manifest() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.get.__name__
command = SmarterJournalCliCommands(command)
metadata = SAMConnectionCommonMetadata(
name="example_connection",
description=f"Example {self.kind} using any of the following authentication methods: {AuthMethods.all()}",
version="0.1.0",
tags=["example", "api", "connection"],
annotations=[
{"smarter.sh/plugin": "example_plugin"},
{"smarter.sh/created_by": "smarter_api_connection_broker"},
],
)
connection = PydanticApiConnection(
baseUrl="http://localhost:9357/",
apiKey="12345-top-secret-67890-fghij",
authMethod="token",
timeout=30,
proxyProtocol="http",
proxyHost="proxy.example.com",
proxyPort=8080,
proxyUsername="proxyuser",
proxyPassword="proxypass",
)
spec = SAMApiConnectionSpec(
connection=connection,
)
status = SAMConnectionCommonStatus(
account_number="2194-1233-0815",
username="admin_user",
recordLocator="abc123def456",
created=datetime.now(),
modified=datetime.now(),
)
sam_api_connection = SAMApiConnection(
apiVersion=self.api_version,
kind=self.kind,
metadata=metadata,
spec=spec,
status=status,
)
# validate our results by round-tripping the data through the Pydantic model
data = json.loads(sam_api_connection.model_dump_json())
return self.json_response_ok(command=command, data=data)
###########################################################################
# Smarter manifest abstract method implementations
###########################################################################
[docs]
def cache_invalidations(self) -> None:
"""
Invalidate any relevant caches when the manifest or connection data changes.
"""
logger.debug("%s.cache_invalidations() called.", self.formatted_class_name_cache_invalidations)
if self.connection:
ApiConnection.get_cached_object(invalidate=True, pk=self.connection.id)
super().cache_invalidations()
[docs]
def get(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Retrieve a list of ApiConnection objects as a journaled JSON response.
This method queries the database for `ApiConnection` instances matching the current account and optional name filter, serializes each result, and returns a structured JSON response including metadata, item count, and model titles.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Optional keyword arguments, such as `name` to filter connections.
:type kwargs: dict
:return: Journaled JSON response containing serialized ApiConnection data.
:rtype: SmarterJournaledJsonResponse
.. seealso::
:class:`ApiConnection`
:class:`ApiConnectionSerializer`
:class:`SmarterJournaledJsonResponse`
:meth:`SAMApiConnectionBroker.SerializerClass`
**Example usage**::
response = broker.get(request)
print(response.data)
# Filter by name
response = broker.get(request, name="my_connection")
print(response.data)
"""
logger.debug(
"%s.get() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.get.__name__
command = SmarterJournalCliCommands(command)
name: Optional[str] = kwargs.get(SAMMetadataKeys.NAME.value, None)
data = []
# generate a QuerySet of ApiConnection objects that match our search criteria
if name:
api_connections = ApiConnection.objects.filter(user_profile__account=self.account, name=name)
else:
api_connections = ApiConnection.objects.filter(user_profile__account=self.account)
model_titles = self.get_model_titles(serializer=self.SerializerClass())
# iterate over the QuerySet and use the manifest controller to create a Pydantic model dump for each ApiConnection
for api_connection in api_connections:
try:
self.connection_init()
self._connection = api_connection
model_dump = self.SerializerClass(api_connection).data
camel_cased_model_dump = self.snake_to_camel(model_dump)
data.append(camel_cased_model_dump)
except Exception as e:
raise SAMConnectionBrokerError(message=str(e), thing=self.kind, command=command) from e
data = {
SAMKeys.APIVERSION.value: self.api_version,
SAMKeys.KIND.value: self.kind,
SAMMetadataKeys.NAME.value: name,
SAMKeys.METADATA.value: {"count": len(data)},
SCLIResponseGet.KWARGS.value: kwargs,
SCLIResponseGet.DATA.value: {
SCLIResponseGetData.TITLES.value: model_titles,
SCLIResponseGetData.ITEMS.value: data,
},
}
return self.json_response_ok(command=command, data=data)
[docs]
def apply(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Apply the manifest. Copy the manifest data to the Django ORM model and
save the model to the database.
This method calls :meth:`super().apply` to ensure that the manifest is loaded
and validated before applying the manifest to the Django ORM model.
Note that there are fields included in the manifest that are not editable
and are therefore removed from the Django ORM model dict prior to attempting
the ``save()`` command. These fields are defined in the ``readonly_fields`` list.
.. note::
tags are handled separately because they are of type TaggableManager and
require a different method to set them.
Example manifest structure::
{
"apiVersion": "smarter.sh/v1", # read only
"kind": "ApiConnection", # read only
"metadata": { # updated in super().apply()
"name": "testf232a0619cb19da0",
"description": "new description",
"version": "1.0.0"
},
"spec": { # updated here.
"connection": {
"kind": "ApiConnection",
"version": "1.0.0",
"account": "2194-1233-0815",
"baseUrl": "http://localhost:9357/api/v1/cli/example_manifest/plugin/",
"apiKey": "testf232a0619cb19da0",
"authMethod": "basic",
"timeout": 30,
"proxyProtocol": "http",
"proxyHost": null,
"proxyPort": null,
"proxyUsername": null,
"proxyPassword": null
}
},
"status": { # read only
"connection_string": "http://localhost:9357/api/v1/cli/example_manifest/plugin/ (Auth: ******)",
"is_valid": false
}
}
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: JSON response indicating success and the updated manifest data.
:rtype: SmarterJournaledJsonResponse
:raises SAMConnectionBrokerError: If an error occurs during update or save.
"""
logger.debug(
"%s.apply() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
super().apply(request, kwargs)
updated = False
command = self.apply.__name__
command = SmarterJournalCliCommands(command)
readonly_fields = ["id", "created_at", "updated_at", "tags"]
if not self.user.is_staff:
raise SAMConnectionBrokerError(
message="Only account admins can apply api connection manifests.",
thing=self.kind,
command=command,
)
# update the spec
api_key_name = camel_to_snake(SAMApiConnectionSpecConnectionKeys.API_KEY.value)
proxy_password_name = camel_to_snake(SAMApiConnectionSpecConnectionKeys.PROXY_PASSWORD.value)
data = self.manifest_to_django_orm()
tags = data.get("tags", [])
for field in readonly_fields:
data.pop(field, None)
try:
for key, value in data.items():
if key == api_key_name:
if self.api_key_secret and key != self.api_key_secret.id: # type: ignore[comparison-overlap]
setattr(self.connection, key, self.api_key_secret)
logger.info("%s.apply() setting api_key Secret <Fk> to %s", self.formatted_class_name, value)
updated = True
elif key == proxy_password_name:
if self.proxy_password_secret and key != self.proxy_password_secret.id: # type: ignore[comparison-overlap]
setattr(self.connection, key, self.proxy_password_secret)
logger.info(
"%s.apply() setting proxy_password Secret <Fk> to %s",
self.formatted_class_name,
value,
)
updated = True
else:
if key != value:
setattr(self.connection, key, value)
logger.info("%s.apply() updating %s to %s", self.formatted_class_name, key, value)
updated = True
if updated and isinstance(self.connection, ApiConnection):
self.connection.save()
self.connection.tags.set(tags)
logger.info(
"%s.apply() updated ApiConnection %s",
self.formatted_class_name,
self.SerializerClass(self.connection).data,
)
except Exception as e:
raise SAMConnectionBrokerError(message=str(e), thing=self.kind, command=command) from e
self.cache_invalidations()
return self.json_response_ok(command=command, data=self.to_json())
[docs]
def chat(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Handle chat operations for the API connection broker.
This method is intended to process chat requests using the manifest broker. Currently, it is not implemented and will always raise a `SAMBrokerErrorNotImplemented` exception.
This method is not implemented. Any invocation will result in an error.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: SAMBrokerErrorNotImplemented. This method always raises an exception.
:rtype: SmarterJournaledJsonResponse
.. seealso::
:class:`SAMApiConnectionBroker`
:class:`SmarterJournalCliCommands`
:class:`SAMBrokerErrorNotImplemented`
**Example usage**::
try:
response = broker.chat(request)
except SAMBrokerErrorNotImplemented as e:
print("Chat not implemented:", e)
"""
logger.debug(
"%s.chat() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.chat.__name__
command = SmarterJournalCliCommands(command)
raise SAMBrokerErrorNotImplemented(message="Chat not implemented", thing=self.kind, command=command)
[docs]
def describe(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Return a JSON response containing the manifest data for the current API connection.
This method serializes the manifest and connection details, including metadata, specification, and status, into a structured JSON response. It validates the connection and includes relevant fields such as connection string and validity status.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: JSON response with manifest data.
:rtype: SmarterJournaledJsonResponse
:raises SAMBrokerErrorNotReady:
If no connection is found
:raises SAMConnectionBrokerError:
if serialization or validation fails.
.. seealso::
:class:`SAMApiConnection`
:class:`ApiConnection`
:class:`SmarterJournaledJsonResponse`
:meth:`SAMApiConnectionBroker.connection`
**Example usage**::
response = broker.describe(request)
print(response.data)
"""
logger.debug(
"%s.describeº() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.describe.__name__
command = SmarterJournalCliCommands(command)
if self.manifest is None:
raise SAMBrokerErrorNotReady(message="No connection found", thing=self.kind, command=command)
try:
data = self.manifest.model_dump()
return self.json_response_ok(command=command, data=data)
except Exception as e:
raise SAMConnectionBrokerError(message=str(e), thing=self.kind, command=command) from e
[docs]
def delete(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Delete the current API connection and return a JSON response indicating the result.
This method attempts to delete the associated `ApiConnection` object from the database. If successful, it returns an empty JSON response. If no connection exists, or if an error occurs during deletion, an appropriate exception is raised.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: JSON response indicating deletion success.
:rtype: SmarterJournaledJsonResponse
:raises SAMBrokerErrorNotReady:
If no connection is found to delete.
:raises SAMConnectionBrokerError:
If an error occurs during deletion.
.. error::
Any exception during deletion is wrapped and raised as :class:`SAMConnectionBrokerError`.
.. seealso::
:class:`ApiConnection`
:class:`SmarterJournaledJsonResponse`
:meth:`SAMApiConnectionBroker.connection`
**Example usage**::
response = broker.delete(request)
print(response.data)
"""
logger.debug(
"%s.delete() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.delete.__name__
command = SmarterJournalCliCommands(command)
if not self.user.is_staff:
raise SAMConnectionBrokerError(
message="Only account admins can delete api connection manifests.",
thing=self.kind,
command=command,
)
if self.connection:
try:
self.connection.delete()
return self.json_response_ok(command=command, data={})
except Exception as e:
raise SAMConnectionBrokerError(message=str(e), thing=self.kind, command=command) from e
raise SAMBrokerErrorNotReady(message="No connection found", thing=self.kind, command=command)
[docs]
def deploy(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Handle deploy operations for the API connection broker.
This is not implemented and will always raise a `SAMBrokerErrorNotImplemented` exception.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: SAMBrokerErrorNotImplemented. This method always raises an exception.
:rtype: SmarterJournaledJsonResponse
"""
logger.debug(
"%s.deploy() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.deploy.__name__
command = SmarterJournalCliCommands(command)
raise SAMBrokerErrorNotImplemented(message="Deploy not implemented", thing=self.kind, command=command)
[docs]
def undeploy(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Handle undeploy operations for the API connection broker.
This is not implemented and will always raise a `SAMBrokerErrorNotImplemented` exception.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: SAMBrokerErrorNotImplemented. This method always raises an exception.
:rtype: SmarterJournaledJsonResponse
"""
logger.debug(
"%s.undeploy() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.undeploy.__name__
command = SmarterJournalCliCommands(command)
raise SAMBrokerErrorNotImplemented(message="Undeploy not implemented", thing=self.kind, command=command)
[docs]
def logs(self, request: "HttpRequest", *args, **kwargs) -> SmarterJournaledJsonResponse:
"""
Handle logs operations for the API connection broker.
This is not implemented and will always raise a `SAMBrokerErrorNotImplemented` exception.
:param request: Django HTTP request object.
:type request: "HttpRequest"
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
:return: SAMBrokerErrorNotImplemented. This method always raises an exception.
:rtype: SmarterJournaledJsonResponse
"""
logger.debug(
"%s.logs() called for %s %s args: %s kwargs: %s",
self.formatted_class_name,
self.kind,
self.name,
args,
kwargs,
)
command = self.logs.__name__
command = SmarterJournalCliCommands(command)
raise SAMBrokerErrorNotImplemented(message="Logs not implemented", thing=self.kind, command=command)