"""Smarter API Manifest - Plugin.spec"""
import os
from typing import Any, ClassVar, List, Optional, Union
from pydantic import Field, field_validator, model_validator
from smarter.apps.plugin.manifest.models.api_plugin.const import MANIFEST_KIND
from smarter.apps.plugin.manifest.models.common import (
Parameter,
RequestHeader,
TestValue,
UrlParam,
)
from smarter.apps.plugin.manifest.models.common.plugin.spec import SAMPluginCommonSpec
from smarter.common.const import SmarterHttpMethods
from smarter.common.exceptions import SmarterValueError
from smarter.lib import logging
from smarter.lib.django.validators import SmarterValidator
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.manifest.exceptions import SAMValidationError
from smarter.lib.manifest.models import SmarterBasePydanticModel
filename = os.path.splitext(os.path.basename(__file__))[0]
MODULE_IDENTIFIER = f"{MANIFEST_KIND}.{filename}"
SMARTER_PLUGIN_MAX_SYSTEM_ROLE_LENGTH = 2048
logger = logging.getSmarterLogger(__name__, any_switches=[SmarterWaffleSwitches.PLUGIN_LOGGING])
[docs]
class ApiData(SmarterBasePydanticModel):
"""Smarter API - apiData class."""
endpoint: str = Field(
...,
max_length=255,
description="The endpoint path for the API. Example: '/v1/weather'.",
)
method: str = Field(
default="GET",
description="The HTTP method to use for the API request. Default is 'GET'.",
max_length=10,
)
urlParams: Optional[List[UrlParam]] = Field(
default=None,
description="A list of URL parameters to be included in the API request. Example: {'city': 'San Francisco'}",
)
headers: Optional[List[RequestHeader]] = Field(
default=None,
description="A list of JSON dict containing headers to be sent with the API request. Example: {'Authorization': 'Bearer <token>'}",
)
body: Optional[Union[dict[str, Any], list[Any]]] = Field(
default=None,
description="Any valid JSON object containing the body of the API request, if applicable.",
)
parameters: Optional[List[Parameter]] = Field(
default=None,
description="A JSON dict containing parameter names and data types. Example: {'city': {'type': 'string', 'description': 'City name'}}",
)
testValues: Optional[List[TestValue]] = Field(
default=None,
description="A JSON dict containing test values for each parameter. Example: {'city': 'San Francisco'}",
)
limit: Optional[int] = Field(
default=100,
gt=1,
description="The maximum number of records to return from the API. Default is 100.",
)
[docs]
@field_validator("endpoint")
def validate_endpoint(cls, v):
try:
SmarterValidator.validate_url_endpoint(v)
except (SAMValidationError, SmarterValueError) as e:
if isinstance(v, str) and not v.endswith("/"):
# Ensure trailing slash
v = v + "/"
try:
SmarterValidator.validate_url_endpoint(v)
except (SAMValidationError, SmarterValueError):
raise SAMValidationError(f"Invalid endpoint: {e}") from e
return v
[docs]
@field_validator("method")
def validate_method(cls, v):
valid_methods = SmarterHttpMethods.all
if v.upper() not in valid_methods:
raise SAMValidationError(f"Invalid HTTP method: {v}. Must be one of {valid_methods}.")
return v.upper()
[docs]
class SAMApiPluginSpec(SAMPluginCommonSpec):
"""Smarter API Manifest ApiConnection.spec"""
class_identifier: ClassVar[str] = MODULE_IDENTIFIER
connection: str = Field(
...,
description=f"{class_identifier}.selector[obj]: the name of an existing ApiConnection to use for the {MANIFEST_KIND}",
)
apiData: ApiData = Field(
..., description=f"{class_identifier}.selector[obj]: the ApiData to use for the {MANIFEST_KIND}"
)
[docs]
@model_validator(mode="after")
def validate_connection(self):
"""
Validate that the connection value is a valid cleanstring and that at
least 1 record exists in the ApiConnection table with the given name.
If the model includes an authenticated user then also validate that at
least 1 record exists in the ApiConnection table with the given name that
is accessible by the authenticated user.
"""
v = self.connection
if not SmarterValidator.is_valid_cleanstring(v):
raise SAMValidationError(f"connection '{v}' must be a valid cleanstring with no illegal characters.")
return self