Source code for cybsi.cloud.iocean.schemas

from typing import Optional

from ..api import Tag
from ..internal import BaseAPI, BaseAsyncAPI, JsonObject, JsonObjectView
from ..pagination import AsyncPage, Cursor, Page
from ..view import _TaggedView

_PATH = "/iocean/schemas"


[docs] class SchemaAPI(BaseAPI): """Schema API."""
[docs] def register(self, schema: JsonObject) -> "SchemaRegistrationView": """Register an object schema. Note: Calls `POST /iocean/schemas`. Args: schema: JSON schema of the object. See :ref:`object_schemas` for information about schema structure. Returns: Schema registration view. Raises: :class:`~cybsi.cloud.error.InvalidRequestError`: Provided values are invalid (see form value requirements). :class:`~cybsi.cloud.error.ConflictError`: Form contains conflict errors. Note: Conflict error codes specific for this method: * :attr:`~cybsi.cloud.error.ConflictErrorCodes.DuplicateSchema` """ resp = self._connector.do_post(path=_PATH, json=schema) return SchemaRegistrationView(resp.json())
[docs] def update( self, *, schema_id: str, tag: Tag, schema: JsonObject, ) -> None: """Update the object schema. Note: Calls `PUT /iocean/schemas/{schema_id}`. Args: schema_id: URL friendly string, uniquely identifies json schema. tag: :attr:`SchemaView.tag` value. Use :meth:`view` to retrieve it. schema: JSON schema of the object. See :ref:`object_schemas` for information about schema structure. Raises: :class:`~cybsi.cloud.error.InvalidRequestError`: Provided values are invalid (see form value requirements). :class:`~cybsi.cloud.error.SemanticError`: Form contains logic errors. :class:`~cybsi.cloud.error.ResourceModifiedError`: Object schema changed since last request. Update tag and retry. :class:`~cybsi.cloud.error.NotFoundError`: Object schema not found. Note: Semantic error codes specific for this method: * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidSchemaID` """ path = f"{_PATH}/{schema_id}" self._connector.do_put(path=path, tag=tag, json=schema)
[docs] def view(self, schema_id: str) -> "SchemaView": """Get the object schema view. Note: Calls `GET /iocean/schemas/{schema_id}`. Args: schema_id: URL friendly string, uniquely identifies json schema. Returns: Schema view. Raises: :class:`~cybsi.cloud.error.NotFoundError`: Object schema not found. """ path = f"{_PATH}/{schema_id}" resp = self._connector.do_get(path=path) return SchemaView(resp)
[docs] def filter( self, *, cursor: Optional[Cursor] = None, limit: Optional[int] = None, ) -> Page["SchemaCommonView"]: """Get an object schemas filtration list. Note: Calls `GET /iocean/schemas`. Args: cursor: Page cursor. limit: Page limit. Returns: Page with schema common views and next page cursor. Raises: :class:`~cybsi.cloud.error.InvalidRequestError`: Provided values are invalid (see form value requirements). """ params: JsonObject = {} if cursor is not None: params["cursor"] = str(cursor) if limit is not None: params["limit"] = limit resp = self._connector.do_get(path=_PATH, params=params) return Page(self._connector.do_get, resp, SchemaCommonView)
[docs] class SchemaAsyncAPI(BaseAsyncAPI): """Schema asynchronous API."""
[docs] async def register(self, schema: JsonObject) -> "SchemaRegistrationView": """Register an object schema. Note: Calls `POST /iocean/schemas`. Args: schema: JSON schema of the object. See :ref:`object_schemas` for information about schema structure. Returns: Schema registration view. Raises: :class:`~cybsi.cloud.error.InvalidRequestError`: Provided values are invalid (see form value requirements). :class:`~cybsi.cloud.error.ConflictError`: Form contains conflict errors. Note: Conflict error codes specific for this method: * :attr:`~cybsi.cloud.error.ConflictErrorCodes.DuplicateSchema` """ resp = await self._connector.do_post(path=_PATH, json=schema) return SchemaRegistrationView(resp.json())
[docs] async def update( self, *, schema_id: str, tag: Tag, schema: JsonObject, ) -> None: """Update the object schema. Note: Calls `PUT /iocean/schemas/{schema_id}`. Args: schema_id: URL friendly string, uniquely identifies json schema. tag: :attr:`SchemaView.tag` value. Use :meth:`view` to retrieve it. schema: JSON schema of the object. See :ref:`object_schemas` for information about schema structure. Raises: :class:`~cybsi.cloud.error.InvalidRequestError`: Provided values are invalid (see form value requirements). :class:`~cybsi.cloud.error.SemanticError`: Form contains logic errors. :class:`~cybsi.cloud.error.ResourceModifiedError`: Object schema changed since last request. Update tag and retry. :class:`~cybsi.cloud.error.NotFoundError`: Object schema not found. Note: Semantic error codes specific for this method: * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidSchemaID` """ path = f"{_PATH}/{schema_id}" await self._connector.do_put(path=path, tag=tag, json=schema)
[docs] async def view(self, schema_id: str) -> "SchemaView": """Get the object schema view. Note: Calls `GET /iocean/schemas/{schema_id}`. Args: schema_id: URL friendly string, uniquely identifies json schema. Returns: Schema view. Raises: :class:`~cybsi.cloud.error.NotFoundError`: Object schema not found. """ path = f"{_PATH}/{schema_id}" resp = await self._connector.do_get(path=path) return SchemaView(resp)
[docs] async def filter( self, *, cursor: Optional[Cursor] = None, limit: Optional[int] = None, ) -> AsyncPage["SchemaCommonView"]: """Get an object schemas filtration list. Note: Calls `GET /iocean/schemas`. Args: cursor: Page cursor. limit: Page limit. Returns: Page with schema common views and next page cursor. Raises: :class:`~cybsi.cloud.error.InvalidRequestError`: Provided values are invalid (see form value requirements). """ params: JsonObject = {} if cursor is not None: params["cursor"] = str(cursor) if limit is not None: params["limit"] = limit resp = await self._connector.do_get(path=_PATH, params=params) return AsyncPage(self._connector.do_get, resp, SchemaCommonView)
[docs] class SchemaRegistrationView(JsonObjectView): """Schema registration view""" @property def schema_id(self) -> str: """URL friendly string, uniquely identifies json schema.""" return self._get("schemaID")
[docs] class SchemaCommonView(JsonObjectView): """Schema common view""" @property def schema_id(self) -> str: """URL friendly string, uniquely identifies json schema.""" return self._get("schemaID") @property def title(self) -> str: """The human-readable name of the json schema.""" return self._get("title")
[docs] class SchemaView(_TaggedView): """Schema view""" @property def schema_id(self) -> str: """URL friendly string, uniquely identifies json schema.""" return self._get("schemaID") @property def schema(self) -> JsonObject: """JSON schema of the object See :ref:`object_schemas` for information about schema structure. """ return self.raw()