Python SDK для лёгкой интеграции с сервисами Cybsi Cloud 
from .__version__ import (  # noqa: F401
    __author__,
    __author_email__,
    __description__,
    __license__,
    __title__,
    __version__,
)

cybsi/__init__.py
cyberstudio/cybsi-cloud-sdk
__version__ = "1.0.4"
__title__ = "cybsi-cloud-sdk"
__description__ = "Cybsi Cloud development kit"
__license__ = "Apache License 2.0"
__author__ = "Cybsi Cloud developers"
__author_email__ = ""

cybsi/__version__.py
cyberstudio/cybsi-cloud-sdk
"""A set of classes and functions implementing low-level Cybsi Cloud API client.

The client allows calling different sections of API.
For example, there's a separate section for authorization,
and a separate section for IOCean.
"""
from .api import Null, Nullable, NullType, Tag

from .client import AsyncClient, Client
from .client_config import Config, Timeouts, Limits

from .enum import CybsiAPIEnum

cybsi/cloud/__init__.py
cyberstudio/cybsi-cloud-sdk
from typing import Optional, TypeVar, Union, cast


class Tag:
    """Identifier of a specific version of a resource.

    Prevents mid-air-collisions.
    Required for edit() functions on Cybsi Cloud API resources.
    """

    pass


# This is a hack to prevent Sphinx autodoc-typehint type inlining.
# If we simply alias Tag = str, it inlines str everywhere,
# and functions lose descriptive parameter and return value types.
# Additionally, this hack prevents SDK users from creating Tag instances.
# Users have to call view()-like methods.
Tag.__supertype__ = str  # type: ignore


class NullType:
    """Depicts explicit `null` values in API.

    The only instance of this type is :data:`Null`.
    """

    pass


Null = NullType()
"""The only possible value of :class:`NullType`"""

T = TypeVar("T")
"""Non-null type."""

Nullable = Union[T, None, NullType]
"""Missing JSON field and ``null`` JSON field sometimes
have a difference in Cybsi Cloud API.
Typically, ``register()`` or ``edit()``-like methods omit unset keyword parameters
when they send request to API.

Nullable is used to indicate optional keyword arguments
which are be passed as explicit ``null`` to API.
"""


# SDK internal function. Placed here for locality with Nullable.
def _unwrap_nullable(value: Nullable[T]) -> Optional[T]:
    return None if value is Null else cast(T, value)

cybsi/cloud/api.py
cyberstudio/cybsi-cloud-sdk
"""Use this section of API operate auth information.
"""
from .api import AuthAPI
from .api_key import (
    APIKeyAuth,
    APIKeysAPI,
    APIKeyRegistrationView,
    APIKeyForm,
    APIKeyView,
)
from .permission import (
    ResourceAction,
    ResourcePermissionView,
    ResourcePermissionForm,
)
from .resource import (
    ResourcesAPI,
    ResourceView,
    ResourceRefView,
)
from .token import TokenType, TokenView

cybsi/cloud/auth/__init__.py
cyberstudio/cybsi-cloud-sdk
from ..internal import BaseAPI
from .api_key import APIKeysAPI
from .resource import ResourcesAPI


class AuthAPI(BaseAPI):
    """Auth API."""

    @property
    def api_keys(self) -> APIKeysAPI:
        """API-Keys API handle."""
        return APIKeysAPI(self._connector)

    @property
    def resources(self) -> ResourcesAPI:
        """Resources API handle."""
        return ResourcesAPI(self._connector)

cybsi/cloud/auth/api.py
cyberstudio/cybsi-cloud-sdk
import asyncio
import json
import logging
import threading
from datetime import datetime
from typing import AsyncGenerator, Generator, Iterable, List, Optional
from urllib.parse import urljoin

import httpx

from ..error import _raise_cybsi_error
from ..internal import (
    BaseAPI,
    JsonObject,
    JsonObjectForm,
    JsonObjectView,
    parse_rfc3339_timestamp,
    rfc3339_timestamp,
)
from ..pagination import Cursor, Page
from .permission import ResourcePermissionForm, ResourcePermissionView
from .token import TokenView

logger = logging.getLogger(__name__)


class APIKeyAuth(httpx.Auth):
    """Automatically handles authentication
    of :class:`~cybsi.cloud.Client` requests using API key.

    Args:
        api_url: Cybsi Cloud auth API URL. Usually equal to Client config API URL.
        api_key: Cybsi Cloud API key.
    """

    requires_response_body = True  # instructs httpx to pass token request response body

    _get_token_path = "/auth/token"

    def __init__(self, *, api_url: str, api_key: str):
        # See https://www.python-httpx.org/advanced/#customizing-authentication
        self._api_key = api_key
        self._api_url = api_url
        self._sync_lock = threading.RLock()
        self._async_lock = asyncio.Lock()
        self._token = ""

    def sync_auth_flow(
        self, request: httpx.Request
    ) -> Generator[httpx.Request, httpx.Response, None]:
        token = self._token
        if token:
            request.headers["Authorization"] = self._token
            response = yield request
            if response.status_code != 401:
                return

        with self._sync_lock:
            if self._token == token:
                token_response = yield self._build_token_request(request)
                self._update_token(token_response, token_response.read())
            request.headers["Authorization"] = self._token
            yield request

    async def async_auth_flow(
        self, request: httpx.Request
    ) -> AsyncGenerator[httpx.Request, httpx.Response]:
        token = self._token
        if token:
            request.headers["Authorization"] = self._token
            response = yield request
            if response.status_code != 401:
                return

        async with self._async_lock:
            if self._token == token:
                token_response = yield self._build_token_request(request)
                self._update_token(token_response, await token_response.aread())
            request.headers["Authorization"] = self._token
            yield request

    def _build_token_request(self, req) -> httpx.Request:
        token_url = urljoin(self._api_url, self._get_token_path)
        headers = {
            "X-Api-Version": req.headers["X-Api-Version"],
            "User-Agent": req.headers["User-Agent"],
            "X-Api-Key": self._api_key,
        }
        return httpx.Request(
            "GET",
            url=token_url,
            headers=headers,
        )

    def _update_token(
        self, token_response: httpx.Response, token_response_content: bytes
    ) -> None:
        if not token_response.is_success:
            _raise_cybsi_error(token_response)
        token = TokenView(json.loads(token_response_content))

        self._token = f"{token.type.value} {token.access_token}"


class APIKeysAPI(BaseAPI):
    """API-Keys API."""

    _path = "/auth/keys"

    def filter(
        self,
        *,
        revoked: Optional[bool] = None,
        description: Optional[str] = None,
        cursor: Optional[Cursor] = None,
    ) -> Page["APIKeyView"]:
        """Get API keys.

        Note:
            Calls `GET /auth/keys`.
        Args:
            revoked: Revocation flag.
            description: Key description.
            cursor: Page cursor.
        Return:
            Page with API-Key common views and next page cursor.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
        """
        params: JsonObject = {}
        if cursor is not None:
            params["cursor"] = str(cursor)
        if revoked is not None:
            params["revoked"] = bool(revoked)
        if description is not None:
            params["description"] = str(description)
        resp = self._connector.do_get(path=self._path, params=params)
        page = Page(self._connector.do_get, resp, APIKeyView)
        return page

    def generate(self, api_key: "APIKeyForm") -> "APIKeyRegistrationView":
        """Generate new API-Key.

        Note:
            Calls `POST /auth/keys`.
        Args:
            api_key: API-Key.
        Returns:
            API-Key view.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.SemanticError`: Form contains logic errors.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.ResourceNotFound`
        """

        resp = self._connector.do_post(path=self._path, json=api_key.json())
        return APIKeyRegistrationView(resp.json())

    def revoke(
        self,
        api_key_id: int,
    ):
        """Revoke API-Key.

        Warning:
            Key revocation is an irreversible operation.
        Note:
            Calls `POST /auth/keys/{api_key_id}/revoked`.
        Args:
            api_key_id: API-Key identifier.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.NotFoundError`: Resource not found.
        """

        path = f"{self._path}/{api_key_id}/revoked"
        self._connector.do_post(path=path)


class APIKeyForm(JsonObjectForm):
    """API-Key form.

    This is the form you need to fill to generate API-Key.

    Args:
        permissions: List of permissions.
        expires_at: Expiration date.
            The API-Key is automatically disabled after the expiration date.
        description: API-Key description.
    """

    def __init__(
        self,
        permissions: Iterable[ResourcePermissionForm],
        *,
        expires_at: Optional[datetime] = None,
        description: Optional[str] = None,
    ):
        super().__init__()
        if expires_at is not None:
            self._data["expiresAt"] = rfc3339_timestamp(expires_at)
        if description is not None:
            self._data["description"] = description
        self._data["permissions"] = [perm.json() for perm in permissions]


class APIKeyRegistrationView(JsonObjectView):
    """API-Key view."""

    @property
    def id(self) -> int:
        """API-Key identifier."""
        return self._get("id")

    @property
    def key(self) -> str:
        """API-Key value.

        Warning:
            Do not forget to save this value. It is not recoverable if lost."""
        return self._get("key")


class APIKeyView(JsonObjectView):
    """API-Key view."""

    @property
    def id(self) -> int:
        """API-Key identifier."""
        return self._get("id")

    @property
    def created_at(self) -> datetime:
        """Creation date."""
        return parse_rfc3339_timestamp(self._get("createdAt"))

    @property
    def expires_at(self) -> Optional[datetime]:
        """Expiration date.
        The API-Key is automatically disabled after the expiration date."""
        return self._map_optional("expiresAt", parse_rfc3339_timestamp)

    @property
    def description(self) -> Optional[str]:
        """API-Key description."""
        return self._get_optional("description")

    @property
    def last_used_at(self) -> Optional[datetime]:
        """Last usage date."""
        return self._map_optional("lastUsedAt", parse_rfc3339_timestamp)

    @property
    def revoked(self) -> bool:
        """API-Key revoked flag."""
        return self._get("revoked")

    @property
    def permissions(self) -> List[ResourcePermissionView]:
        """List of permissions."""
        return [ResourcePermissionView(perm) for perm in self._get("permissions")]

cybsi/cloud/auth/api_key.py
cyberstudio/cybsi-cloud-sdk
from typing import Iterable, List

from enum_tools import document_enum

from ..enum import CybsiAPIEnum
from ..internal import JsonObjectForm, JsonObjectView
from .resource import ResourceRefView


@document_enum
class ResourceAction(CybsiAPIEnum):
    """Resource action."""

    Read = "Read"
    """Allows to read the resource."""
    Write = "Write"
    """Allows to modify the resource."""
    ReadChildren = "ReadChildren"
    """Allows to read direct children resources without direct permission."""
    WriteChildren = "WriteChildren"
    """Allows to modify direct children resources without direct permission."""


class ResourcePermissionView(JsonObjectView):
    """Resource permission."""

    @property
    def resource(self) -> ResourceRefView:
        """Resource."""
        return ResourceRefView(self._get("resource"))

    @property
    def actions(self) -> List[ResourceAction]:
        """List of permitted actions."""
        return [ResourceAction(act) for act in self._get("actions")]


class ResourcePermissionForm(JsonObjectForm):
    """Resource permissions form.

    This is the form for resource permissions you have to fill to generate API-Key.

    Args:
        resource_id: resource identifier.
        actions: permitted actions for resource.
    """

    def __init__(
        self,
        *,
        resource_id: int,
        actions: Iterable[ResourceAction],
    ):
        super().__init__()
        self._data["resource"] = {"id": resource_id}
        self._data["actions"] = [act.value for act in actions]

cybsi/cloud/auth/permission.py
cyberstudio/cybsi-cloud-sdk
from typing import Optional

from ..error import JsonObject
from ..internal import BaseAPI, JsonObjectView
from ..pagination import Cursor, Page


class ResourcesAPI(BaseAPI):
    """Resources API."""

    _path = "/auth/resources"

    def filter(
        self,
        *,
        parent_id: Optional[int] = None,
        cursor: Optional[Cursor] = None,
    ) -> Page["ResourceView"]:
        """Get resources.

        Note:
            Calls `GET /auth/resources`.
        Args:
            parent_id: identifier of parent resource. It must be greater than 0.
            cursor: Page cursor.
        Return:
            Page with resource common views and next page cursor.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.SemanticError`: Form contains logic errors.
        Note:
            Semantic error codes specific for this method:
            * :attr:`~cybsi.cloud.error.SemanticErrorCodes.ResourceNotFound`
        """
        params: JsonObject = {}
        if parent_id is not None:
            params["parentID"] = parent_id
        if cursor is not None:
            params["cursor"] = str(cursor)
        resp = self._connector.do_get(path=self._path, params=params)
        page = Page(self._connector.do_get, resp, ResourceView)
        return page


class ResourceRefView(JsonObjectView):
    """Resource reference view."""

    @property
    def id(self) -> int:
        """API-Key identifier."""
        return self._get("id")


class ResourceView(JsonObjectView):
    """Resource view."""

    @property
    def id(self) -> int:
        """Resource identifier."""
        return self._get("id")

    @property
    def name(self) -> str:
        """Resource name."""
        return self._get("name")

    @property
    def parent(self) -> Optional[ResourceRefView]:
        """Parent resource."""
        return self._map_optional("parent", ResourceRefView)

cybsi/cloud/auth/resource.py
cyberstudio/cybsi-cloud-sdk
import datetime

from enum_tools import document_enum

from cybsi.cloud.enum import CybsiAPIEnum

from ..internal import JsonObjectView


@document_enum
class TokenType(CybsiAPIEnum):
    """Token type."""

    Bearer = "Bearer"  # doc: Authentication is made using Bearer token type


class TokenView(JsonObjectView):
    """Authorization token view."""

    @property
    def access_token(self) -> str:
        """Access token. A cryptic string."""
        return self._get("accessToken")

    @property
    def type(self) -> TokenType:
        """Token type."""
        return TokenType(self._get("tokenType"))

    @property
    def expires_in(self) -> datetime.timedelta:
        """Token lifetime until expiration."""
        seconds = int(self._get("expiresIn"))
        return datetime.timedelta(seconds=seconds)

cybsi/cloud/auth/token.py
cyberstudio/cybsi-cloud-sdk
from .auth import APIKeyAuth, AuthAPI
from .client_config import Config
from .internal import AsyncHTTPConnector, HTTPConnector
from .iocean import IOCeanAPI, IOCeanAsyncAPI


class Client:
    """The main entry point for all actions with Cybsi Cloud REST API.

    As the client is low-level, it is structured around Cybsi Cloud REST API routes.
    Use properties of the client to retrieve handles of API sections.

    The client also follows Cybsi Cloud REST API input-output formats,
    providing little to no abstration from JSONs API uses.
    It's relatively easy to construct an invalid request,
    so use client's functions wisely.

    Hint:
        Use :class:`~cybsi.cloud.Client` properties
        to construct needed API handles. Don't construct sub-APIs manually.

        Do this:
            >>> from cybsi.cloud import Client
            >>> client = Client(config)
            >>> res = client.iocean.collections.filter()
        Not this:
            >>> from cybsi.cloud.iocean import IOCeanAPI
            >>> IOceanAPI(connector).collections

    Args:
        config: Client config.
    Usage:
        >>> from cybsi.cloud import Config, Client
        >>> api_key = "8Nqjk6V4Q_et_Rf5EPu4SeWy4nKbVPKPzKJESYdRd7E"
        >>> config = Config(api_key)
        >>> client = Client(config)
        >>>
        >>> collections = client.iocean.collections.filter()
        >>> print(collections.data())
        >>> client.close()  # "with" syntax is also supported for Client
    """

    def __init__(self, config: Config):
        auth = APIKeyAuth(api_url=config.api_url, api_key=config.api_key)

        self._connector = HTTPConnector(
            base_url=config.api_url,
            auth=auth,
            ssl_verify=config.ssl_verify,
            timeouts=config.timeouts,
            limits=config.limits,
        )

    def __enter__(self) -> "Client":
        self._connector.__enter__()
        return self

    def __exit__(
        self,
        exc_type=None,
        exc_value=None,
        traceback=None,
    ) -> None:
        self._connector.__exit__(exc_type, exc_value, traceback)

    def close(self) -> None:
        """Close client and release connections."""
        self._connector.close()

    @property
    def auth(self) -> AuthAPI:
        """Auth API handle."""
        return AuthAPI(self._connector)

    @property
    def iocean(self) -> IOCeanAPI:
        """IOCean API handle."""
        return IOCeanAPI(self._connector)


class AsyncClient:
    """The asynchronous analog of :class:`Client`.

    As you can see, the asynchronous client has fewer features than synchronous one.
    This is because we don't simply copy-paste features,
    but provide them only when they're actually useful in asynchronous applications.

    Args:
        config: Client config.
    """

    def __init__(self, config: Config):
        auth = APIKeyAuth(api_url=config.api_url, api_key=config.api_key)

        self._connector = AsyncHTTPConnector(
            base_url=config.api_url,
            auth=auth,
            ssl_verify=config.ssl_verify,
            timeouts=config.timeouts,
            limits=config.limits,
        )

    async def __aenter__(self) -> "AsyncClient":
        await self._connector.__aenter__()
        return self

    async def __aexit__(
        self,
        exc_type=None,
        exc_value=None,
        traceback=None,
    ) -> None:
        await self._connector.__aexit__(exc_type, exc_value, traceback)

    async def aclose(self) -> None:
        """Close client and release connections."""
        await self._connector.aclose()

    @property
    def iocean(self) -> IOCeanAsyncAPI:
        """IOCean asynchronous API handle."""
        return IOCeanAsyncAPI(self._connector)

cybsi/cloud/client.py
cyberstudio/cybsi-cloud-sdk
from typing import Optional, Union

import httpx

from .api import Null, Nullable, _unwrap_nullable

TimeoutTypes = Union[
    Optional[float],
    "Timeouts",
]


class Limits:
    """Configuration for limits to various client behaviors.

    Args:
        max_connections: The maximum number of concurrent connections that may be
            established.
        max_keepalive_connections: Allow the connection pool to maintain
            keep-alive connections below this point. Should be less than or equal
            to `max_connections`.
        keepalive_expiry: The maximum time (in second) to allow before closing
            a keep-alive connection.
    """

    def __init__(
        self,
        *,
        max_connections: Optional[int] = None,
        max_keepalive_connections: Optional[int] = None,
        keepalive_expiry: Optional[float] = 5.0,
    ):
        self.max_connections = max_connections
        self.max_keepalive_connections = max_keepalive_connections
        self.keepalive_expiry = keepalive_expiry

    def _as_httpx_limits(self) -> httpx.Limits:
        return httpx.Limits(
            max_connections=self.max_connections,
            max_keepalive_connections=self.max_keepalive_connections,
            keepalive_expiry=self.keepalive_expiry,
        )


class Timeouts:
    """Timeout configuration.

    Note:
        Default parameters value is :data:`~cybsi.cloud.Null`,
        it means "use default timeout".
        :data:`None` value means "infinite timeout".

        Please note those timeouts don't restrict total execution time
        of SDK methods (like register, view and so on).
        Wrap calls to SDK methods with asyncio.wait_for or
        similar functions if you want to control total execution time.
    Args:
        default: Timeout (in seconds) on all operations listed below.
        connect: The maximum amount of time (in seconds) to wait
            for a connection attempt to a server to succeed.
        read: The maximum amount of time (in seconds) to wait between
            consecutive read operations for a response from the server.
        write: The maximum amount of time (in seconds) to wait between
            consecutive write operations (send request) to the server.
    Raises:
        :class:`ValueError`: Timeout must either default settings
            for all operations or set all three parameters explicitly.
    Usage:
        >>> # No timeouts.
        >>> Timeouts(default=None)
        >>> # 5s timeout on all operations.
        >>> Timeouts(default=5.0)
        >>> # 5s timeout on connect, no other timeouts.
        >>> Timeouts(default=None, connect=5.0)
        >>> # 10s timeout on connect. 5s timeout elsewhere.
        >>> Timeouts(default=5.0, connect=10.0)
    """

    def __init__(
        self,
        *,
        default: Nullable[TimeoutTypes] = Null,
        connect: Nullable[float] = Null,
        read: Nullable[float] = Null,
        write: Nullable[float] = Null,
    ):
        if isinstance(default, Timeouts):
            # Passed as a single explicit Timeout.
            assert connect is Null
            assert read is Null
            assert write is Null
            self.connect = _unwrap_nullable(default.connect)  # type: Optional[float]
            self.read = _unwrap_nullable(default.read)  # type: Optional[float]
            self.write = _unwrap_nullable(default.write)  # type: Optional[float]

        elif not (connect is Null or read is Null or write is Null):
            self.connect = _unwrap_nullable(connect)
            self.read = _unwrap_nullable(read)
            self.write = _unwrap_nullable(write)
        else:
            if default is Null:
                raise ValueError(
                    "Timeout must either include default settings "
                    "for all operations or set all three parameters explicitly."
                )
            self.connect = (
                _unwrap_nullable(default)
                if connect is Null
                else _unwrap_nullable(connect)
            )
            self.read = (
                _unwrap_nullable(default) if read is Null else _unwrap_nullable(read)
            )
            self.write = (
                _unwrap_nullable(default) if write is Null else _unwrap_nullable(write)
            )

    def _as_httpx_timeouts(self) -> httpx.Timeout:
        return httpx.Timeout(
            connect=self.connect,
            read=self.read,
            write=self.write,
            # Timeout (in seconds) on acquiring connection from pool.
            # Internal default httpx.Timeout setting.
            pool=5.0,
        )


DEFAULT_TIMEOUTS = Timeouts(default=60.0)
DEFAULT_LIMITS = Limits(max_connections=100, max_keepalive_connections=20)


class Config:
    """:class:`Client` config.

    Args:
        api_key: Cybsi Cloud API key.
        api_url: Base API URL.
        ssl_verify: Enable SSL certificate verification.
        timeouts: Timeout configuration. Default configuration is 60 sec
            on all operations.
        limits:  Configuration for limits to various client behaviors.
            Default configuration is max_connections=100, max_keepalive_connections=20.
    """

    def __init__(
        self,
        *,
        api_key: str,
        api_url: str = "https://cybsi.cloud",
        ssl_verify: bool = True,
        timeouts: Timeouts = DEFAULT_TIMEOUTS,
        limits: Limits = DEFAULT_LIMITS,
    ):
        self.api_key = api_key
        self.api_url = api_url
        self.ssl_verify = ssl_verify
        self.timeouts = timeouts
        self.limits = limits

cybsi/cloud/client_config.py
cyberstudio/cybsi-cloud-sdk
from enum import Enum
from functools import lru_cache
from typing import TypeVar

ET = TypeVar("ET")


class CybsiAPIEnum(Enum):
    """CybsiAPIEnum is a base class for all Cybsi Cloud API enumerations."""

    @classmethod
    @lru_cache()
    def from_string(cls, value: str, ignore_case=False):
        """Convert a string value to enumeration value.

        Args:
            value: value to convert.
            ignore_case: ignore enumeration values case.
        Return:
            Enumeration value.
        """
        try:
            return cls(value)
        except ValueError as exp:
            if not ignore_case:
                raise exp from None

            for member in cls:
                if str(member.value).lower() == value.lower():
                    return member

            raise exp from None

cybsi/cloud/enum.py
cyberstudio/cybsi-cloud-sdk
"""This section documents exceptions SDK can raise.

Those are exceptions SDK client can expect if SDK is used in correct way.
If your Python code didn't respect the expected model, there's no guarantees.
For example, if function argument types are ignored,
SDK can raise exceptions not listed here.

Each exception type is annotated if it makes sense to retry.

Some exceptions have ``code`` property. It allows to determine the concrete error.
"""
from typing import Any, Dict, Optional, cast

import httpx
from enum_tools import document_enum

from .enum import CybsiAPIEnum


class CybsiError(Exception):
    """Base exception used by SDK. Sometimes can be retried.

    If it's not of one of subclasses (see below), means unexpected error.

    :class:`CybsiError` covers those cases:

    * Unexpected API response status code. See :class:`APIError` for specific errors.
    * Unexpected API response content.
    * Connection error.

    """

    def __init__(self, message: str, ex: Optional[Exception] = None):
        if ex and str(ex) != "":
            message = f"{message}: {ex}"
        super().__init__(message)
        self._ex = ex


JsonObject = Dict[str, Any]


class APIError(CybsiError):
    """Base exception for HTTP 4xx API responses."""

    def __init__(
        self,
        status: int,
        content: JsonObject,
        header: Optional[str] = None,
        suffix: Optional[str] = None,
    ) -> None:
        self._status = status
        self._view = ErrorView(content)
        self._header = (
            header
            if header is not None
            else f"API response error. HTTP status: {self._status}"
        )
        self._suffix = (
            suffix
            if suffix is not None
            else f"code: {self._view.code}, message: {self._view.message}"
        )

        msg = self._header
        if self._suffix:
            msg += f", {self._suffix}"
        super().__init__(msg)

    @property
    def content(self) -> "ErrorView":
        return self._view


class InvalidRequestError(APIError):
    """Invalid request error. Retry will never work.

    Ideally, should not be raised by SDK. If it's raised, it means one of two things:

    * SDK was used incorrectly. For example, values of invalid type
      were passed to SDK functions.
    * There's a bug in request serialization in SDK. Please report the bug.

    """

    BadRequest = "BadRequest"
    InvalidData = "InvalidData"
    InvalidPathArgument = "InvalidPathArgument"
    InvalidQueryArgument = "InvalidQueryArgument"

    def __init__(self, content: JsonObject) -> None:
        super().__init__(400, content, header="invalid request")


class UnauthorizedError(APIError):
    """Client lacks valid authentication credentials. Retry will never work."""

    Unauthorized = "Unauthorized"

    def __init__(self, content: JsonObject) -> None:
        super().__init__(401, content, header="operation not authorized")


class ForbiddenError(APIError):
    """Operation was forbidden. Retry will not work unless system is reconfigured."""

    def __init__(self, content: JsonObject) -> None:
        super().__init__(403, content, header="operation forbidden")

    @property
    def code(self) -> "ForbiddenErrorCodes":
        """Error code."""
        return ForbiddenErrorCodes(self._view.code)


class NotFoundError(APIError):
    """Requested resource not found. Retry will never work."""

    def __init__(self, content: JsonObject) -> None:
        super().__init__(404, {}, header="resource not found", suffix="")


class ConflictError(APIError):
    """Resource already exists. Retry will never work."""

    def __init__(self, content: JsonObject) -> None:
        super().__init__(409, content, header="resource already exists")

    @property
    def code(self) -> "ConflictErrorCodes":
        """Error code."""
        return ConflictErrorCodes(self._view.code)


class ResourceModifiedError(APIError):
    """Resource was modified since last read. **Retry is a must**.

    Read the updated resource from API, and apply your modifications again.
    """

    def __init__(self, content: JsonObject) -> None:
        super().__init__(
            412, content, header="resource was modified since last read", suffix=""
        )


class RequestEntityTooLargeError(APIError):
    """Request content is too large."""

    def __init__(self, content: JsonObject) -> None:
        super().__init__(413, content, header="request content is too large", suffix="")


class SemanticError(APIError):
    """Semantic error. Retry will not work (almost always).

    Request syntax was valid, but system business rules forbid the request.

    For example, we're trying to unpack an artifact, but the artifact is not an archive.
    """

    def __init__(self, content: JsonObject) -> None:
        super().__init__(422, content, header="semantic error")

    @property
    def code(self) -> "SemanticErrorCodes":
        """Error code."""
        return SemanticErrorCodes(self._view.code)


@document_enum
class ForbiddenErrorCodes(CybsiAPIEnum):
    """Possible error codes of :class:`ForbiddenError`."""

    MissingPermissions = "MissingPermissions"
    """Authenticated but not authorized to perform operation."""
    Forbidden = "Forbidden"
    """Other cases."""


@document_enum
class ConflictErrorCodes(CybsiAPIEnum):
    """Conflict error codes."""

    DuplicateSchema = "DuplicateSchema"
    """Schema with the given schemaID is already registered."""
    DuplicateCollection = "DuplicateCollection"
    """Collection with the given name is already registered."""


@document_enum
class SemanticErrorCodes(CybsiAPIEnum):
    """Semantic error codes."""

    ResourceNotFound = "ResourceNotFound"
    """Cloud resource not found."""
    InvalidSchemaID = "InvalidSchemaID"
    """schemaID parameter can't be changed."""
    SchemaNotFound = "SchemaNotFound"
    """The specified schema is not found"""

    # Objects
    InvalidKeyFormat = "InvalidKeyFormat"
    """Object key has invalid format."""
    InvalidKeySet = "InvalidKeySet"
    """Object has invalid key set."""
    KeySetConflict = "KeySetConflict"
    """Object key intersects with another one."""
    SchemaCheckFail = "SchemaCheckFail"
    """Object validation by schema failed."""
    CursorOutOfRange = "CursorOutOfRange"
    """Cursor for collection changes is obsolete."""


class ErrorView(dict):
    """Error returned by Cybsi Cloud API."""

    @property
    def code(self) -> str:
        """Error code."""

        return cast(str, self.get("code"))

    @property
    def message(self) -> str:
        """Error message."""

        return cast(str, self.get("message"))

    @property
    def details(self) -> JsonObject:
        """Error details."""

        return cast(JsonObject, self.get("details"))


class SchemaCheckErrorDetails(dict):
    """Details for schema check error."""

    @property
    def absolute_keyword_location(self) -> str:
        """Absolute validation path of validating schema."""

        return cast(str, self.get("AbsoluteKeywordLocation"))

    @property
    def instance_location(self) -> str:
        """Location of the json value within the instance being validated."""

        return cast(str, self.get("InstanceLocation"))

    @property
    def message(self) -> str:
        """Error description."""

        return cast(str, self.get("Message"))


_error_mapping = {
    400: InvalidRequestError,
    401: UnauthorizedError,
    403: ForbiddenError,
    404: NotFoundError,
    409: ConflictError,
    412: ResourceModifiedError,
    413: RequestEntityTooLargeError,
    422: SemanticError,
}


def _raise_cybsi_error(resp: httpx.Response) -> None:
    err_cls = _error_mapping.get(resp.status_code, None)
    if err_cls is not None:
        raise err_cls(resp.json())

    raise CybsiError(
        f"unexpected response status code: {resp.status_code}. "
        f"Request body: {resp.text}"
    )

cybsi/cloud/error.py
cyberstudio/cybsi-cloud-sdk
from .base import (
    BaseAPI,
    BaseAsyncAPI,
    JsonObject,
    JsonObjectForm,
    JsonObjectView,
    list_mapper,
)
from .connector import AsyncHTTPConnector, HTTPConnector
from .time import (
    parse_rfc3339_timestamp,
    rfc3339_timestamp,
)

cybsi/cloud/internal/__init__.py
cyberstudio/cybsi-cloud-sdk
"""
Base internal classes, useful to simplify API implementation.
"""

import json
from typing import Any, Callable, Dict, List, Optional, TypeVar

from ..error import CybsiError
from .connector import AsyncHTTPConnector, HTTPConnector

JsonObject = Dict[str, Any]


class BaseAPI:
    # Base class for all API handle implementations.
    def __init__(self, connector: HTTPConnector):
        self._connector = connector


class BaseAsyncAPI:
    # Base class for all async API handle implementations.
    def __init__(self, connector: AsyncHTTPConnector):
        self._connector = connector


class JsonObjectForm:
    def __init__(self, data: Optional[JsonObject] = None):
        self._data = data or {}

    def __str__(self):
        return json.dumps(self._data, indent=2)

    def json(self):
        return self._data


class JsonObjectView:
    def __init__(self, data: Optional[JsonObject] = None):
        self._data = data or {}

    def __str__(self):
        return json.dumps(self._data, indent=2)

    def _get(self, key):
        try:
            return self._data[key]
        except KeyError as exp:
            msg = f"{self.__class__.__name__} does not have field: {exp}"
            raise CybsiError(msg) from None

    def _get_optional(self, key):
        return self._data.get(key, None)

    def _map_optional(self, key, mapper: Callable[[Any], Any]):
        value = self._get_optional(key)
        if value is not None:
            return mapper(value)
        return None

    def _map_list_optional(self, key, mapper: Callable[[Any], Any]):
        values = self._get_optional(key)
        if values is not None:
            return [mapper(val) for val in values]
        return None

    def raw(self) -> JsonObject:
        """Returns raw object view"""
        return self._data


T = TypeVar("T")


def list_mapper(item_creator: Callable[..., T]) -> Callable[..., List[T]]:
    def _create_typed_list(items: List) -> List[T]:
        return [item_creator(item) for item in items]

    return _create_typed_list

cybsi/cloud/internal/base.py
cyberstudio/cybsi-cloud-sdk
from typing import Any, Optional

import httpx

from cybsi.__version__ import __version__

from ..api import Tag
from ..error import CybsiError, _raise_cybsi_error

_BASIC_HEADERS = {
    "User-Agent": f"cybsi-cloud-client/v{__version__}",
    "X-Api-Version": "1",
}

_IF_MATCH_HEADER = "If-Match"


class HTTPConnector:
    """Connector performing round trips to Cybsi Cloud."""

    def __init__(
        self,
        base_url: str,
        auth: Any,
        ssl_verify: bool,
        timeouts,
        limits,
    ):
        self._client = httpx.Client(
            auth=auth,
            verify=ssl_verify,
            base_url=base_url,
            headers=_BASIC_HEADERS,
            timeout=timeouts._as_httpx_timeouts(),
            limits=limits._as_httpx_limits(),
        )

    def __enter__(self) -> "HTTPConnector":
        self._client.__enter__()
        return self

    def __exit__(
        self,
        exc_type=None,
        exc_value=None,
        traceback=None,
    ) -> None:
        self._client.__exit__(exc_type, exc_value, traceback)

    def close(self) -> None:
        """Close client and release connections."""
        self._client.close()

    def do_get(
        self, path: str, params: Optional[dict] = None, stream=False, **kwargs
    ) -> httpx.Response:
        if params is None:
            params = {}
        return self._do("GET", path, params=params, stream=stream, **kwargs)

    def do_post(self, path: str, json: Any = None, **kwargs) -> httpx.Response:
        return self._do("POST", path, json=json, **kwargs)

    def do_patch(
        self, path: str, tag: Tag, json: Any = None, **kwargs
    ) -> httpx.Response:
        headers = kwargs.setdefault("headers", {})
        headers[_IF_MATCH_HEADER] = tag
        return self._do("PATCH", path, json=json, **kwargs)

    def do_put(
        self, path: str, tag: Optional[Tag] = None, json: Any = None, **kwargs
    ) -> httpx.Response:
        if tag is not None:
            headers = kwargs.setdefault("headers", {})
            headers[_IF_MATCH_HEADER] = tag
        return self._do("PUT", path, json=json, **kwargs)

    def do_delete(
        self, path: str, params: Optional[dict] = None, **kwargs
    ) -> httpx.Response:
        return self._do("DELETE", path, params=params, **kwargs)

    def _do(self, method: str, path: str, stream=False, **kwargs):
        """Do HTTP request.

        Args:
            method: HTTP method i.e GET, POST, PUT.
            path: URL path.
            kwargs: Any kwargs supported by httpx.Request.
        Return:
            Response.
        Raise:
            :class:`~cybsi.cloud.error.CybsiError`: On connectivity issues.
            :class:`~cybsi.cloud.error.APIError`: If response status code is >= 400
        """
        req = self._client.build_request(method, url=path, **kwargs)
        try:
            resp = self._client.send(request=req, stream=stream)
        except CybsiError:
            raise
        except httpx.TimeoutException as exp:
            # Handle httpx.TimeoutException separately
            # because it doesn't have an exception message.
            raise CybsiError("request timeout", exp) from exp
        except Exception as exp:
            raise CybsiError("could not send request", exp) from exp

        if not resp.is_success:
            if resp.stream:  # type: ignore
                # read stream data to raise the error
                resp.read()
            _raise_cybsi_error(resp)

        return resp


class AsyncHTTPConnector:
    """Asynchronous connector performing round trips to Cybsi Cloud."""

    def __init__(
        self,
        base_url: str,
        auth: Any,
        ssl_verify: bool,
        timeouts,
        limits,
    ):
        self._client = httpx.AsyncClient(
            auth=auth,
            verify=ssl_verify,
            base_url=base_url,
            headers=_BASIC_HEADERS,
            timeout=timeouts._as_httpx_timeouts(),
            limits=limits._as_httpx_limits(),
        )

    async def __aenter__(self) -> "AsyncHTTPConnector":
        await self._client.__aenter__()
        return self

    async def __aexit__(
        self,
        exc_type=None,
        exc_value=None,
        traceback=None,
    ) -> None:
        await self._client.__aexit__(exc_type, exc_value, traceback)

    async def aclose(self) -> None:
        """Close client and release connections."""
        await self._client.aclose()

    async def do_get(
        self, path: str, params: Optional[dict] = None, stream=False, **kwargs
    ) -> httpx.Response:
        if params is None:
            params = {}
        return await self._do("GET", path, params=params, stream=stream, **kwargs)

    async def do_post(self, path: str, json=None, **kwargs) -> httpx.Response:
        return await self._do("POST", path, json=json, **kwargs)

    async def do_patch(
        self, path: str, tag: Tag, json: Any = None, **kwargs
    ) -> httpx.Response:
        headers = kwargs.setdefault("headers", {})
        headers[_IF_MATCH_HEADER] = tag
        return await self._do("PATCH", path, json=json, **kwargs)

    async def do_put(
        self, path: str, tag: Optional[Tag] = None, json=None, **kwargs
    ) -> httpx.Response:
        if tag is not None:
            headers = kwargs.setdefault("headers", {})
            headers[_IF_MATCH_HEADER] = tag
        return await self._do("PUT", path, json=json, **kwargs)

    async def do_delete(
        self, path: str, params: Optional[dict] = None, **kwargs
    ) -> httpx.Response:
        return await self._do("DELETE", path, params=params, **kwargs)

    async def _do(self, method: str, path: str, stream=False, **kwargs):
        """Do HTTP request.

        Args:
            method: HTTP method i.e GET, POST, PUT.
            path: URL path.
            kwargs: Any kwargs supported by httpx.Request.
        Return:
            Response.
        Raise:
            :class:`~cybsi.cloud.error.CybsiError`: On connectivity issues.
            :class:`~cybsi.cloud.error.APIError`: If response status code is >= 400
        """
        req = self._client.build_request(method, url=path, **kwargs)
        try:
            resp = await self._client.send(request=req, stream=stream)
        except CybsiError:
            raise
        except httpx.TimeoutException as exp:
            # Handle httpx.TimeoutException separately
            # because it doesn't have an exception message.
            raise CybsiError("request timeout", exp) from exp
        except Exception as exp:
            raise CybsiError("could not send request", exp) from exp

        if not resp.is_success:
            if resp.stream:  # type: ignore
                # read stream data to raise the error
                await resp.aread()
            _raise_cybsi_error(resp)

        return resp

cybsi/cloud/internal/connector.py
cyberstudio/cybsi-cloud-sdk
import datetime


def rfc3339_timestamp(dt: datetime.datetime) -> str:
    """
    Generate an :RFC:`3339`-formatted timestamp in UTC time zone from a
    :class:`datetime.datetime`.
    Microseconds are ignored.
    >>> import datetime as dtm
    >>> rfc3339_timestamp(dtm.datetime(2009,1,1,12,59,59,0,dtm.timezone.utc))
    '2009-01-01T12:59:59Z'
    If timestamp is naive, local time zone is used to calculate shift.
    >>> rfc3339_timestamp(dtm.datetime(2009,1,1,12,59,59,0))
    '2009-01-01T06:59:59Z'
    """
    dt = dt.astimezone(datetime.timezone.utc)
    timestamp = dt.strftime("%Y-%m-%dT%H:%M:%SZ")
    return timestamp


def parse_rfc3339_timestamp(ts: str) -> datetime.datetime:
    if ts.find(".") != -1:
        return datetime.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%S.%fZ")
    else:
        return datetime.datetime.strptime(ts, "%Y-%m-%dT%H:%M:%SZ")

cybsi/cloud/internal/time.py
cyberstudio/cybsi-cloud-sdk
"""Use this section of API to access IOCean objects, collections and schemas.
"""

from .api import IOCeanAPI, IOCeanAsyncAPI
from .collection import (
    CollectionAPI,
    CollectionAsyncAPI,
    CollectionForm,
    CollectionCommonView,
    CollectionRegistrationView,
    CollectionView,
)
from .objects import (
    ObjectAPI,
    ObjectsAsyncAPI,
    ObjectKeyType,
    ObjectType,
    ObjectView,
    ObjectKeyView,
    ObjectChangeView,
    ObjectOperation,
)
from .schemas import (
    SchemaAPI,
    SchemaAsyncAPI,
    SchemaView,
    SchemaCommonView,
    SchemaRegistrationView,
)

cybsi/cloud/iocean/__init__.py
cyberstudio/cybsi-cloud-sdk
from ..internal import BaseAPI, BaseAsyncAPI
from .collection import CollectionAPI, CollectionAsyncAPI
from .objects import ObjectAPI, ObjectsAsyncAPI
from .schemas import SchemaAPI, SchemaAsyncAPI


class IOCeanAPI(BaseAPI):
    """IOCean API."""

    @property
    def collections(self) -> CollectionAPI:
        """Get IOCean collections handle."""
        return CollectionAPI(self._connector)

    @property
    def schemas(self) -> SchemaAPI:
        """Get IOCean schemas handle."""
        return SchemaAPI(self._connector)

    @property
    def objects(self) -> ObjectAPI:
        """Objects API handle."""
        return ObjectAPI(self._connector)


class IOCeanAsyncAPI(BaseAsyncAPI):
    """IOCean asynchronous API."""

    @property
    def collections(self) -> CollectionAsyncAPI:
        """Collections asynchronous API handle."""
        return CollectionAsyncAPI(self._connector)

    @property
    def schemas(self) -> SchemaAsyncAPI:
        """Schemas asynchronous API handle."""
        return SchemaAsyncAPI(self._connector)

    @property
    def objects(self) -> ObjectsAsyncAPI:
        """Objects asynchronous API handle."""
        return ObjectsAsyncAPI(self._connector)

cybsi/cloud/iocean/api.py
cyberstudio/cybsi-cloud-sdk
from typing import Optional

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

_PATH = "/iocean/collections"


class CollectionAPI(BaseAPI):
    """Collection API"""

    def register(self, collection: "CollectionForm") -> "CollectionRegistrationView":
        """Register collection.

        Note:
            Calls `POST /iocean/collections`.
        Args:
            collection: collection form.
        Return:
            Collection registration view.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.ConflictError`:
                Collection with the specified id (name) already exists.
            :class:`~cybsi.cloud.error.SemanticError`:
                Form contains logic errors.
        Note:
            Conflict error codes specific for this method:
            * :attr:`~cybsi.cloud.error.ConflictErrorCodes.DuplicateCollection`
            Semantic error codes specific for this method:
            * :attr:`~cybsi.cloud.error.SemanticErrorCodes.SchemaNotFound`
        """
        resp = self._connector.do_post(_PATH, json=collection.json())
        return CollectionRegistrationView(resp.json())

    def view(self, collection_id: str) -> "CollectionView":
        """Get collection.

        Note:
            Calls `GET /iocean/collections/{collectionName}`.
        Args:
            collection_id: collection's id.
        Return:
            Collection view.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`:
                Resource not found.
        """

        url = f"{_PATH}/{collection_id}"
        resp = self._connector.do_get(url)
        return CollectionView(resp)

    def view_schema(self, collection_id: str) -> SchemaView:
        """Get collection schema.

        Note:
            Calls `GET /iocean/collections/{collectionName}/schema`.
        Args:
            collection_id: collection's id.
        Return:
            Schema view.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`:
                Resource not found.
        """

        url = f"{_PATH}/{collection_id}/schema"
        resp = self._connector.do_get(url)
        return SchemaView(resp)

    def update(self, collection_id: str, tag: Tag, *, schema_id: Optional[str]):
        """Update collection

        Note:
            Calls `PATCH /iocean/collections/{collectionName}`.
        Args:
            collection_id: collection's id.
            tag: :attr:`CollectionView.tag` value. Use :meth:`view` to retrieve it.
            schema_id: schema identified.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see form value requirements).
            :class:`~cybsi.cloud.error.ResourceModifiedError`:
                Object schema changed since last request. Update tag and retry.
            :class:`~cybsi.cloud.error.NotFoundError`:
                Resource not found.
            :class:`~cybsi.cloud.error.SemanticError`:
                Form contains logic errors.
        Note:
            Semantic error codes specific for this method:
            * :attr:`~cybsi.cloud.error.SemanticErrorCodes.SchemaNotFound`
        """
        body = {}
        if schema_id is not None:
            body["schemaID"] = schema_id

        url = f"{_PATH}/{collection_id}"
        self._connector.do_patch(url, tag=tag, json=body)

    def filter(
        self, *, cursor: Optional[Cursor] = None
    ) -> Page["CollectionCommonView"]:
        """Get collections.

        Note:
            Calls `GET /iocean/collections`.
        Args:
            cursor: Page cursor.
        Return:
            Page with collection views and next page cursor.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
        """
        params: JsonObject = {}
        if cursor is not None:
            params["cursor"] = str(cursor)
        resp = self._connector.do_get(path=_PATH, params=params)
        return Page(self._connector.do_get, resp, CollectionCommonView)


class CollectionAsyncAPI(BaseAsyncAPI):
    """Collection asynchronous API"""

    async def register(
        self, collection: "CollectionForm"
    ) -> "CollectionRegistrationView":
        """Register collection.

        Note:
            Calls `POST /iocean/collections`.
        Args:
            collection: collection form.
        Return:
            Collection registration view.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.ConflictError`:
                Collection with the specified id (name) already exists.
            :class:`~cybsi.cloud.error.SemanticError`:
                Form contains logic errors.
        Note:
            Conflict error codes specific for this method:
            * :attr:`~cybsi.cloud.error.ConflictErrorCodes.DuplicateCollection`
            Semantic error codes specific for this method:
            * :attr:`~cybsi.cloud.error.SemanticErrorCodes.SchemaNotFound`
        """
        resp = await self._connector.do_post(_PATH, json=collection.json())
        return CollectionRegistrationView(resp.json())

    async def view(self, collection_id: str) -> "CollectionView":
        """Get collection.

        Note:
            Calls `GET /iocean/collections/{collectionName}`.
        Args:
            collection_id: collection's id.
        Return:
            Collection view.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`:
                Resource not found.
        """

        url = f"{_PATH}/{collection_id}"
        resp = await self._connector.do_get(url)
        return CollectionView(resp)

    async def view_schema(self, collection_id: str) -> SchemaView:
        """Get collection schema.

        Note:
            Calls `GET /iocean/collections/{collectionName}/schema`.
        Args:
            collection_id: collection's id.
        Return:
            Schema view.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`:
                Resource not found.
        """

        url = f"{_PATH}/{collection_id}/schema"
        resp = await self._connector.do_get(url)
        return SchemaView(resp)

    async def update(self, collection_id: str, tag: Tag, *, schema_id: Optional[str]):
        """Update collection

        Note:
            Calls `PATCH /iocean/collections/{collectionName}`.
        Args:
            collection_id: collection's id.
            tag: :attr:`CollectionView.tag` value. Use :meth:`view` to retrieve it.
            schema_id: schema identified.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see form value requirements).
            :class:`~cybsi.cloud.error.ResourceModifiedError`:
                Object schema changed since last request. Update tag and retry.
            :class:`~cybsi.cloud.error.NotFoundError`:
                Resource not found.
            :class:`~cybsi.cloud.error.SemanticError`:
                Form contains logic errors.
        Note:
            Semantic error codes specific for this method:
            * :attr:`~cybsi.cloud.error.SemanticErrorCodes.SchemaNotFound`
        """
        body = {}
        if schema_id is not None:
            body["schemaID"] = schema_id

        url = f"{_PATH}/{collection_id}"
        await self._connector.do_patch(url, tag=tag, json=body)

    async def filter(
        self, *, cursor: Optional[Cursor] = None
    ) -> AsyncPage["CollectionCommonView"]:
        """Get collections.

        Note:
            Calls `GET /iocean/collections`.
        Args:
            cursor: Page cursor.
        Return:
            Page with collection views and next page cursor.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
        """
        params: JsonObject = {}
        if cursor is not None:
            params["cursor"] = str(cursor)
        resp = await self._connector.do_get(path=_PATH, params=params)
        return AsyncPage(self._connector.do_get, resp, CollectionCommonView)


class CollectionRegistrationView(JsonObjectView):
    """Collection registration view"""

    @property
    def name(self):
        """The name of registered collection"""
        return self._get("name")


class CollectionCommonView(JsonObjectView):
    """Collection common view"""

    @property
    def name(self) -> str:
        """Collection name"""
        return self._get("name")

    @property
    def schema_id(self) -> str:
        """Schema identifier"""
        return self._get("schemaID")


class CollectionView(_TaggedView, CollectionCommonView):
    """Collection view"""

    pass


class CollectionForm(JsonObjectForm):
    """Collection form"""

    def __init__(self, name: str, schema_id: str):
        super().__init__()
        self._data["name"] = name
        self._data["schemaID"] = schema_id

cybsi/cloud/iocean/collection.py
cyberstudio/cybsi-cloud-sdk
from typing import Iterable, List, Optional, Tuple, cast
from urllib.parse import parse_qs, urlparse

import httpx
from enum_tools import document_enum

from ..enum import CybsiAPIEnum
from ..internal import BaseAPI, BaseAsyncAPI, JsonObject, JsonObjectView
from ..pagination import AsyncPage, Cursor, Page


@document_enum
class ObjectKeyType(CybsiAPIEnum):
    """Object key type."""

    MD5Hash = "MD5Hash"
    SHA1Hash = "SHA1Hash"
    SHA256Hash = "SHA256Hash"
    SHA512Hash = "SHA512Hash"
    DomainName = "DomainName"
    URL = "URL"
    IPAddress = "IPAddress"
    IPNetwork = "IPNetwork"


@document_enum
class ObjectType(CybsiAPIEnum):
    """Object type."""

    File = "File"
    DomainName = "DomainName"
    URL = "URL"
    IPAddress = "IPAddress"
    IPNetwork = "IPNetwork"


_PATH = "/iocean/collections/{}/objects"


@document_enum
class ObjectOperation(CybsiAPIEnum):
    """Object change operation."""

    Add = "Add"
    """Object was added to collection."""
    Remove = "Remove"
    """Object war removed from collection."""
    Update = "Update"
    """Object in collection was updated."""


class ObjectAPI(BaseAPI):
    """Object API."""

    def add(
        self,
        *,
        collection_id: str,
        obj_type: ObjectType,
        keys: Iterable[Tuple[ObjectKeyType, str]],
        context: JsonObject = {},
    ) -> None:
        """Add object to collection.

        If there is registered object with corresponding keys
        and there are no keys conflicts, this method:
        - rewrites object context with new one;
        - extends key set of the registered object.

        Note:
            Calls `POST /iocean/collections/{collection_id}/objects`.
        Args:
            collection_id: Collection identifier.
            obj_type: Type of the object.
            keys: Keys of the object.
            context: Additional data describing object.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.SemanticError`: Request contains logic errors.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidKeyFormat`
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidKeySet`
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.KeySetConflict`
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.SchemaCheckFail`
        """
        path = _PATH.format(collection_id)
        payload = {
            "type": obj_type.value,
            "keys": [{"type": key[0].value, "value": key[1]} for key in keys],
            "context": context,
        }
        self._connector.do_post(path=path, json=payload)

    def delete(
        self,
        *,
        collection_id: str,
        key_type: ObjectKeyType,
        key_value: str,
    ) -> None:
        """Delete object from collection.

        Note:
            Calls `DELETE /iocean/collections/{collection_id}/objects`.
        Args:
            collection_id: Collection identifier.
            key_type: Key type of object to be removed.
            key_value: Key value of object to be removed.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.SemanticError`: Request contains logic errors.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidKeyFormat`
        """
        params: JsonObject = {
            "objectKeyType": key_type.value,
            "objectKey": key_value,
        }
        path = _PATH.format(collection_id)
        self._connector.do_delete(path=path, params=params)

    def filter(
        self, *, collection_id: str, cursor: Optional[Cursor] = None
    ) -> Tuple[Page["ObjectView"], Optional[Cursor]]:
        """Get objects from the collection.

        Note:
            Calls `GET /iocean/collections/{collection_id}/objects`.
        Args:
            collection_id: Collection identifier.
            cursor: Page cursor.
        Return:
            Page with object views. The page contains next page cursor.
            Changes cursor. The cursor can be used to call :meth:`changes`.
        Note:
            Changes cursor is returned only on the first page.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
        """
        params: JsonObject = {}
        if cursor is not None:
            params["cursor"] = str(cursor)
        path = _PATH.format(collection_id)
        resp = self._connector.do_get(path=path, params=params)
        return Page(self._connector.do_get, resp, ObjectView), _extract_changes_cursor(
            resp
        )

    def changes(
        self, *, collection_id: str, cursor: Cursor
    ) -> Page["ObjectChangeView"]:
        """Get objects changes from the collection.

        Note:
            Calls `GET /iocean/collections/{collection_id}/objects/changes`.
        Args:
            collection_id: Collection identifier.
            cursor: Page cursor.
                On the first request you should pass the cursor value
                obtained when requesting objects :meth:`filter`.
                Subsequent calls should use cursor property of the page
                returned by :meth:`changes`.
        Return:
            Page with changes.
        Warning:
            Cursor behaviour differs from other API methods.

            Do not save returned page cursor if it is :data:`None`.
            :data:`None` means that all changes **for this moment** are received.
            More changes can arrive later. Pass your previous non-none ``cursor``
            value in loop, until :data:`None` cursor is returned.

            Please wait some time if method returns a page with :data:`None` cursor.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.SemanticError`: Semantic request error.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.CursorOutOfRange`
        """
        params: JsonObject = {"cursor": cursor}
        path = _PATH.format(collection_id) + "/changes"
        resp = self._connector.do_get(path=path, params=params)
        return Page(self._connector.do_get, resp, ObjectChangeView)


class ObjectsAsyncAPI(BaseAsyncAPI):
    """Object asynchronous API."""

    async def add(
        self,
        *,
        collection_id: str,
        obj_type: ObjectType,
        keys: Iterable[Tuple[ObjectKeyType, str]],
        context: JsonObject = {},
    ) -> None:
        """Add object to collection.

        If there is registered object with corresponding keys
        and there are no keys conflicts, this method:
        - rewrites object context with new one;
        - extends key set of the registered object.

        Note:
            Calls `POST /iocean/collections/{collection_id}/objects`.
        Args:
            collection_id: Collection identifier.
            obj_type: Type of the object.
            keys: Keys of the object.
            context: Additional data describing object.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.SemanticError`: Request contains logic errors.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidKeyFormat`
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidKeySet`
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.KeySetConflict`
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.SchemaCheckFail`
        """
        path = _PATH.format(collection_id)
        payload = {
            "type": obj_type.value,
            "keys": [{"type": key[0].value, "value": key[1]} for key in keys],
            "context": context,
        }
        await self._connector.do_post(path=path, json=payload)

    async def delete(
        self,
        *,
        collection_id: str,
        key_type: ObjectKeyType,
        key_value: str,
    ) -> None:
        """Delete object from collection.

        Note:
            Calls `DELETE /iocean/collections/{collection_id}/objects`.
        Args:
            collection_id: Collection identifier.
            key_type: Key type of object to be removed.
            key_value: Key value of object to be removed.
        Raises:
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.SemanticError`: Request contains logic errors.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.InvalidKeyFormat`
        """
        params: JsonObject = {
            "objectKeyType": key_type.value,
            "objectKey": key_value,
        }
        path = _PATH.format(collection_id)
        await self._connector.do_delete(path=path, params=params)

    async def filter(
        self, *, collection_id: str, cursor: Optional[Cursor] = None
    ) -> Tuple[AsyncPage["ObjectView"], Optional[Cursor]]:
        """Get objects from the collection.

        Note:
            Calls `GET /iocean/collections/{collection_id}/objects`.
        Args:
            collection_id: Collection identifier.
            cursor: Page cursor.
        Return:
            Page with object views. The page contains next page cursor.
            Changes cursor. The cursor can be used to call :meth:`changes`.
        Note:
            Changes cursor is returned only on the first page.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
        """
        params: JsonObject = {}
        if cursor is not None:
            params["cursor"] = str(cursor)
        path = _PATH.format(collection_id)
        resp = await self._connector.do_get(path=path, params=params)
        return AsyncPage(
            self._connector.do_get, resp, ObjectView
        ), _extract_changes_cursor(resp)

    async def changes(
        self, *, collection_id: str, cursor: Cursor
    ) -> AsyncPage["ObjectChangeView"]:
        """Get objects changes from the collection.

        Note:
            Calls `GET /iocean/collections/{collection_id}/objects/changes`.
        Args:
            collection_id: Collection identifier.
            cursor: Page cursor.
                On the first request you should pass the cursor value
                obtained when requesting objects :meth:`filter`.
                Subsequent calls should use cursor property of the page
                returned by :meth:`changes`.
        Return:
            Page with changes.
        Warning:
            Cursor behaviour differs from other API methods.

            Do not save returned page cursor if it is :data:`None`.
            :data:`None` means that all changes **for this moment** are received.
            More changes can arrive later. Pass your previous non-none ``cursor``
            value in loop, until :data:`None` cursor is returned.

            Please wait some time if method returns a page with :data:`None` cursor.
        Raises:
            :class:`~cybsi.cloud.error.NotFoundError`: Collection not found.
            :class:`~cybsi.cloud.error.InvalidRequestError`:
                Provided values are invalid (see args value requirements).
            :class:`~cybsi.cloud.error.SemanticError`: Semantic request error.
        Note:
            Semantic error codes specific for this method:
              * :attr:`~cybsi.cloud.error.SemanticErrorCodes.CursorOutOfRange`
        """
        params: JsonObject = {"cursor": cursor}
        path = _PATH.format(collection_id) + "/changes"
        resp = await self._connector.do_get(path=path, params=params)
        return AsyncPage(self._connector.do_get, resp, ObjectChangeView)


def _extract_changes_cursor(resp: httpx.Response) -> Optional[Cursor]:
    """Extracts changes cursor from response."""
    related_url = resp.links.get("related", {}).get("url")
    if related_url is None:
        return None
    parsed = urlparse(related_url)
    query = parse_qs(parsed.query)
    cursor = query["cursor"]
    return cast(Optional[Cursor], cursor[0]) if cursor is not None else None


class ObjectKeyView(JsonObjectView):
    """Object key view"""

    @property
    def type(self) -> ObjectKeyType:
        """Object key type"""
        return ObjectKeyType.from_string(self._get("type"))

    @property
    def value(self) -> str:
        """Key value."""
        return self._get("value")


class ObjectView(JsonObjectView):
    """Object view."""

    @property
    def type(self) -> ObjectType:
        """Object type."""
        return ObjectType.from_string(self._get("type"))

    @property
    def keys(self) -> List[ObjectKeyView]:
        """Object keys."""
        return [ObjectKeyView(key) for key in self._get("keys")]

    @property
    def context(self) -> JsonObject:
        """Object context."""
        return cast(JsonObject, self._get("context"))


class ObjectChangeView(JsonObjectView):
    """Object change view."""

    @property
    def operation(self) -> ObjectOperation:
        """Object operation."""
        return ObjectOperation.from_string(self._get("operation"))

    @property
    def obj(self) -> ObjectView:
        """Object."""
        return ObjectView(self._get("object"))

cybsi/cloud/iocean/objects.py
cyberstudio/cybsi-cloud-sdk
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"


class SchemaAPI(BaseAPI):
    """Schema API."""

    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())

    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)

    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)

    def filter(
        self,
        *,
        cursor: Optional[Cursor] = None,
    ) -> Page["SchemaCommonView"]:
        """Get an object schemas filtration list.

        Note:
            Calls `GET /iocean/schemas`.
        Args:
            cursor: Page cursor.
        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)
        resp = self._connector.do_get(path=_PATH, params=params)
        return Page(self._connector.do_get, resp, SchemaCommonView)


class SchemaAsyncAPI(BaseAsyncAPI):
    """Schema asynchronous API."""

    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())

    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)

    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)

    async def filter(
        self,
        *,
        cursor: Optional[Cursor] = None,
    ) -> AsyncPage["SchemaCommonView"]:
        """Get an object schemas filtration list.

        Note:
            Calls `GET /iocean/schemas`.
        Args:
            cursor: Page cursor.
        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)
        resp = await self._connector.do_get(path=_PATH, params=params)
        return AsyncPage(self._connector.do_get, resp, SchemaCommonView)


class SchemaRegistrationView(JsonObjectView):
    """Schema registration view"""

    @property
    def schema_id(self) -> str:
        """URL friendly string, uniquely identifies json schema."""
        return self._get("schemaID")


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")


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()

cybsi/cloud/iocean/schemas.py
cyberstudio/cybsi-cloud-sdk
"""
Pagination API.

See Also:
    See :ref:`pagination-example`
    for complete examples of pagination usage.
"""
from typing import (
    AsyncIterator,
    Callable,
    Coroutine,
    Generic,
    Iterator,
    List,
    Optional,
    TypeVar,
    cast,
)
from urllib.parse import parse_qs, urlparse

import httpx


class Cursor:
    """Page cursor value.

    If :data:`None` cursor value is **passed** to SDK
    (as a parameter to filter-like functions), SDK retrieves the first page.

    Typically, SDK does not **return** cursors as function return values.
    SDK returns :class:`Page`, and
    cursor is a property of the page.

    :data:`None` cursor value of a page means last page.
    """

    pass


# This is a hack to prevent Sphinx autodoc-typehint type inlining.
# If we simply alias Cursor = str, it inlines str everywhere,
# and functions lose descriptive parameter and return value types.
# Additionally, this hack prevents SDK users from creating Cursor instances.
# Users have to call filter()-like methods.
Cursor.__supertype__ = str  # type: ignore

T = TypeVar("T")


class _BasePage(Generic[T]):
    def __init__(self, resp: httpx.Response, view: Callable[..., T]):
        self._resp = resp
        self._view = view

    @property
    def next_link(self) -> str:
        """Next page link."""
        return cast(str, self._resp.links.get("next", {}).get("url"))

    @property
    def cursor(self) -> Optional[Cursor]:
        """Page cursor. The current position in the collection.

        :data:`None` means the page is last one.
        """
        next_url = self._resp.links.get("next", {}).get("url")
        if next_url is None:
            return None
        parsed = urlparse(next_url)
        query = parse_qs(parsed.query)
        cursor = query.get("cursor")
        return cast(Optional[Cursor], cursor[0]) if cursor is not None else None

    def data(self) -> List[T]:
        """Get page data as a list of items."""
        return list(iter(self))

    def __iter__(self) -> Iterator[T]:
        yield from (self._view(x) for x in self._resp.json())


class Page(_BasePage[T]):
    """Page returned by Cybsi Cloud API.
       Should not be constructed manually, use filter-like methods provided by SDK.

    Args:
        api_call: Callable object for getting next page
        resp: Response which represents a start page
        view: View class for page elements
    """

    def __init__(
        self,
        api_call: Callable[..., httpx.Response],
        resp: httpx.Response,
        view: Callable[..., T],
    ):
        super().__init__(resp, view)
        self._api_call = api_call

    def next_page(self) -> "Optional[Page[T]]":
        """Get next page.
        If there is no link to the next page it return None.
        """
        if self.next_link is None:
            return None

        return Page(self._api_call, self._api_call(self.next_link), self._view)


class AsyncPage(_BasePage[T]):
    """Page returned by Cybsi API.
       Should not be constructed manually, use filter-like methods provided by SDK.

    Args:
        api_call: Callable object for getting next page
        resp: Response which represents a start page
        view: View class for page elements
    """

    def __init__(
        self,
        api_call: Callable[..., Coroutine],
        resp: httpx.Response,
        view: Callable[..., T],
    ):
        super().__init__(resp, view)
        self._api_call = api_call

    async def next_page(self) -> "Optional[AsyncPage[T]]":
        """Get next page.
        If there is no link to the next page it return None.
        """
        if self.next_link is None:
            return None
        resp = await self._api_call(self.next_link)
        return AsyncPage(self._api_call, resp, self._view)


def chain_pages(start_page: Page[T]) -> Iterator[T]:
    """Get chain of collection objects."""

    page: Optional[Page[T]] = start_page
    while page:
        yield from page
        page = page.next_page()


async def chain_pages_async(start_page: AsyncPage[T]) -> AsyncIterator[T]:
    """Get chain of collection objects asynchronously."""
    page: Optional[AsyncPage[T]] = start_page
    while page:
        for elem in page:
            yield elem
        page = await page.next_page()

cybsi/cloud/pagination.py
cyberstudio/cybsi-cloud-sdk
from typing import cast

import httpx

from .api import Tag
from .internal import JsonObjectView


class _TaggedView(JsonObjectView):
    _etag_header = "ETag"

    def __init__(self, resp: httpx.Response):
        super().__init__(resp.json())
        self._tag = cast(Tag, resp.headers.get(self._etag_header, ""))

    @property
    def tag(self) -> Tag:
        """Resource tag.

        Protects against concurrent object changes.
        Alternatively, can be used for caching.
        """
        return self._tag

cybsi/cloud/view.py
cyberstudio/cybsi-cloud-sdk
# -*- coding: utf-8 -*-
# Cybsi Cloud SDK documentation build configuration file.

import sys
import os

# Insert Cybsi Cloud SDK path into the system.
sys.path.insert(0, os.path.abspath(".."))
sys.path.insert(0, os.path.abspath("_themes"))

import cybsi  # noqa: E402


# -- General configuration ------------------------------------------------
extensions = [
    "sphinx.ext.autodoc",
    "sphinx.ext.napoleon",
    "sphinx_autodoc_typehints",
    "sphinx.ext.intersphinx",
    "sphinx.ext.todo",
    "sphinx.ext.viewcode",
    "enum_tools.autoenum",
]

# A list of regular expressions that match anchors Sphinx should skip
# when checking the validity of anchors in links.
# https://www.sphinx-doc.org/en/master/usage/configuration.html#confval-linkcheck_anchors_ignore
linkcheck_anchors_ignore = [
    "/artifacts/browse/tree/General/cybsi-cloud-pypi"  # anchor of link to pt artifactory
]

templates_path = ["_templates"]

# The suffix(es) of source filenames.
source_suffix = ".rst"

# The master toctree document.
master_doc = "index"

# General information about the project.
project = u"Cybsi Cloud SDK"
copyright = u"Cybsi Cloud developers"
author = u"Cybsi Cloud developers"
version = cybsi.__version__
release = cybsi.__version__
language = "en"

# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
exclude_patterns = ["_build"]

# If true, '()' will be appended to :func: etc. cross-reference text.
add_function_parentheses = True

# If true, the current module name will be prepended to all description
# unit titles (such as .. function::).
add_module_names = True

# Warn about all references where the target cannot be found.
nitpicky = True
nitpick_ignore = [
    ("py:class", "T"),  # Ignore classic typevar
    ("py:obj", "cybsi.cloud.api.T"),  # https://github.com/sphinx-doc/sphinx/issues/9705
    ("py:obj", "cybsi.cloud.pagination.T"),  # https://github.com/sphinx-doc/sphinx/issues/9705
    ("py:class", "httpx.Response"),
    ("py:class", "enum.EnumMeta"),
    ("py:class", "typing.AbstractContextManager"),
    ("py:class", "typing.AbstractAsyncContextManager"),
]

# The name of the Pygments (syntax highlighting) style to use.
pygments_style = "colorful"

# If true, `todo` and `todoList` produce output, else they produce nothing.
todo_include_todos = True

# -- Options for HTML output ----------------------------------------------

# Docs of sphinx_book_theme https://pydata-sphinx-theme.readthedocs.io/en/latest/user_guide/install.html
html_theme = 'sphinx_book_theme'
html_theme_options = {
    "repository_url": "https://github.com/cyberstudio/cybsi-cloud-sdk",
    "use_repository_button": True,
    "use_issues_button": True,
    'secondary_sidebar_items': [],
}
html_static_path = ["_static"]

# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
html_use_smartypants = False

# Custom sidebar templates, maps document names to template names.
html_sidebars = {
    "index": ["sidebar.html", "sourcelink.html", "searchbox.html"],
    "**": [
        "sidebar.html",
        "localtoc.html",
        "relations.html",
        "sourcelink.html",
        "searchbox.html",
    ],
}

# If true, links to the reST sources are added to the pages.
html_show_sourcelink = False

# If true, "Created using Sphinx" is shown in the HTML footer. Default is True.
html_show_sphinx = False

# If true, "(C) Copyright ..." is shown in the HTML footer. Default is True.
html_show_copyright = True

html_search_language = 'en'

# Output file base name for HTML help builder.
htmlhelp_basename = "cybsisdkdoc"

# -- Options for LaTeX output ---------------------------------------------

# Grouping the document tree into LaTeX files. List of tuples
# (source start file, target name, title,
#  author, documentclass [howto, manual, or own class]).
latex_documents = [
    (master_doc, "cybsi.tex", u"Cybsi Cloud SDK Documentation",
     author, "manual")
]

# LaTeX document customization
latex_elements = {
    # \\usepackage[X2,T1]{fontenc} if you need Cyrillic letters (Cybsi работает)
    'fontenc': '\\usepackage[T1]{fontenc}'
}

# -- Options for manual page output ---------------------------------------

# One entry per manual page. List of tuples
# (source start file, name, description, authors, manual section).
man_pages = [
    (master_doc, "cybsi_sdk", u"Cybsi Cloud SDK Documentation",
     [author], 1)]

# -- Options controlling document structure --------------------------------
autodoc_member_order = 'bysource'


# -- Options for type hints in developer documentation ---------------------
intersphinx_mapping = {
    "python": ("https://docs.python.org/3/", None),
}

autodoc_typehints = 'none'
napoleon_use_param = True
napoleon_use_rtype = True
napoleon_preprocess_types = True
napoleon_attr_annotations = True
napoleon_use_ivar = True
typehints_document_rtype = True
typehints_fully_qualified = False
always_document_param_types = False

docs/conf.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from cybsi.cloud import Client, Config
from cybsi.cloud.error import (
    ConflictError,
    SchemaCheckErrorDetails,
    SemanticError,
    SemanticErrorCodes,
)
from cybsi.cloud.iocean.objects import ObjectKeyType, ObjectType

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")
    client = Client(config)

    collection_id = "example-collection"
    keys = [(ObjectKeyType.MD5Hash, "cea239ce075fcb2151ce9ee10227f042")]
    context = {"size": 112}
    try:  # add collection object
        client.iocean.objects.add(
            collection_id=collection_id,
            obj_type=ObjectType.File,
            keys=keys,
            context=context,
        )
    except ConflictError:
        # handle Duplicate Error here
        exit(1)
    except SemanticError as ex:
        if ex.code == SemanticErrorCodes.SchemaCheckFail:
            details = SchemaCheckErrorDetails(ex.content.details)
            print(details.absolute_keyword_location)
            print(details.instance_location)
            print(details.message)
        exit(1)

    client.close()

examples/add_collection_object.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
import asyncio

from cybsi.cloud import AsyncClient, Config
from cybsi.cloud.iocean.objects import ObjectKeyType, ObjectType


async def main():
    config = Config(api_key="the cryptic string")

    collection_id = "example-collection"
    object_files = (
        (
            ObjectType.File,
            [(ObjectKeyType.MD5Hash, "cea239ce075fcb2151ce9ee10227f042")],
            {"size": 112},
        ),
        (
            ObjectType.File,
            [(ObjectKeyType.MD5Hash, "cea239ce075fcb2151ce9ee10227f032")],
            {"size": 10},
        ),
        (
            ObjectType.File,
            [(ObjectKeyType.MD5Hash, "cea239ce075fcb2151ce9ee10227f022")],
            {"size": 1113},
        ),
        (
            ObjectType.File,
            [(ObjectKeyType.MD5Hash, "cea239ce075fcb2151ce9ee10227f012")],
            {"size": 333},
        ),
    )

    async with AsyncClient(config) as client:
        objects = [
            client.iocean.objects.add(
                collection_id=collection_id,
                obj_type=typ,
                keys=keys,
                context=context,
            )
            for typ, keys, context in object_files
        ]
        await asyncio.gather(*objects)


if __name__ == "__main__":
    asyncio.run(main())

examples/add_collection_object_async.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from cybsi.cloud import Client, Config
from cybsi.cloud.pagination import chain_pages

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")

    with Client(config) as client:
        start_page = client.auth.api_keys.filter()
        for item in chain_pages(start_page):
            # Do something with an item
            pass

examples/auth_api_keys_chained.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from datetime import datetime, timedelta

from cybsi.cloud import Client, Config
from cybsi.cloud.auth import APIKeyForm, ResourceAction, ResourcePermissionForm

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")

    with Client(config) as client:
        permissions = [
            ResourcePermissionForm(resource_id=1, actions=[ResourceAction.Read])
        ]
        expires_at = datetime.now() + timedelta(hours=9)
        new_key = APIKeyForm(
            permissions, expires_at=expires_at, description="new test key"
        )
        key_ref = client.auth.api_keys.generate(new_key)
        # do something with the key
        # do not forget to save key_ref.key. It is not recoverable if lost.

examples/auth_generate_api_key.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from cybsi.cloud import Client, Config
from cybsi.cloud.pagination import chain_pages

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")

    with Client(config) as client:
        start_page = client.auth.resources.filter()
        for item in chain_pages(start_page):
            # Do something with an item
            pass

examples/auth_resources_chained.py
cyberstudio/cybsi-cloud-sdk
from cybsi.cloud import Client, Config, Limits, Timeouts

if __name__ == "__main__":
    # Set custom timeouts and limits of HTTP client
    limits = Limits(max_connections=100, max_keepalive_connections=20)
    timeouts = Timeouts(default=3.0)
    config = Config(
        api_key="the cryptic string", ssl_verify=False, timeouts=timeouts, limits=limits
    )
    client = Client(config)
    # then use client as usual

examples/client_advanced_config.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from cybsi.cloud import Client, Config
from cybsi.cloud.iocean.objects import ObjectKeyType

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")
    client = Client(config)

    collection_id = "example-collection"
    client.iocean.objects.delete(
        collection_id=collection_id,
        key_type=ObjectKeyType.MD5Hash,
        key_value="cea239ce075fcb2151ce9ee10227f042",
    )

    client.close()

examples/delete_collection_object.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from cybsi.cloud import Client, Config
from cybsi.cloud.pagination import chain_pages

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")

    with Client(config) as client:
        collection_id = "example-collection"
        start_page, _ = client.iocean.objects.filter(
            collection_id=collection_id,
        )
        for item in chain_pages(start_page):
            # Do something with an item
            pass

examples/get_collection_objects_chained.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
import time
from typing import Iterable, Optional

from cybsi.cloud import Client, Config
from cybsi.cloud.iocean import ObjectChangeView
from cybsi.cloud.pagination import Cursor, Page


def main():
    config = Config(api_key="the cryptic string")

    with Client(config) as client:
        collection_id = "example-collection"
        _, changes_cursor = client.iocean.objects.filter(
            collection_id=collection_id,
        )

        process_changes(client, collection_id, changes_cursor)


def process_changes(client: Client, collection_id: str, changes_cursor: Cursor):
    fetcher = ChangesFetcher(client, collection_id, changes_cursor)
    while True:
        handle_changes(fetcher.changes())
        # Do not forget to persist last changes cursor
        changes_cursor = fetcher.cursor

        time.sleep(10)


class ChangesFetcher:
    def __init__(self, client: Client, collection_id: str, changes_cursor: Cursor):
        self._client = client
        self._collection_id = collection_id
        self._changes_cursor = changes_cursor

    def changes(self) -> Iterable[ObjectChangeView]:
        """Lazily fetch all available collection changes."""
        page: Optional[Page[ObjectChangeView]] = self._client.iocean.objects.changes(
            collection_id=self._collection_id, cursor=self._changes_cursor
        )

        while page:
            # changes page may return empty cursor.
            # Do not save empty cursor to prevent last cursor lost.
            if page.cursor is not None:
                self._changes_cursor = page.cursor
            yield from page
            page = page.next_page()

    @property
    def cursor(self):
        """Current cursor required to fetch next changes page."""
        return self._changes_cursor


def handle_changes(changes: Iterable[ObjectChangeView]):
    """Handle collection changes."""
    cnt = 0
    for item in changes:
        # Do something with item
        cnt += 1
        pass
    print(f"handled {cnt} changes")


if __name__ == "__main__":
    main()

examples/get_collection_objects_changes.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
from typing import Optional

from cybsi.cloud import Client, Config
from cybsi.cloud.iocean import CollectionCommonView
from cybsi.cloud.pagination import Page

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")

    with Client(config) as client:
        page: Optional[Page[CollectionCommonView]] = client.iocean.collections.filter()
        while page:
            # Page is iterable
            for item in page:
                # Do something with an item
                pass
            # Fetch next page
            page = page.next_page()

examples/pagination_manual.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
import asyncio
from hashlib import md5

from cybsi.cloud import AsyncClient, Config
from cybsi.cloud.iocean.objects import ObjectKeyType, ObjectType


async def main():
    config = Config(api_key="the cryptic string")

    collection_id = "example-collection"
    async with AsyncClient(config) as client:
        objects = (
            client.iocean.objects.add(
                collection_id=collection_id,
                obj_type=ObjectType.File,
                keys=[(ObjectKeyType.MD5Hash, md5(str(i).encode("ascii")).hexdigest())],
                context={"size": i + 1},
            )
            for i in range(1000)
        )
        await asyncio.gather(*objects)


if __name__ == "__main__":
    asyncio.run(main())

examples/push_collection_objects.py
cyberstudio/cybsi-cloud-sdk
#!/usr/bin/env python3
import json

from cybsi.cloud import Client, Config
from cybsi.cloud.error import ConflictError

if __name__ == "__main__":
    config = Config(api_key="the cryptic string")
    client = Client(config)

    jsonSchema = """
    {
      "$schema": "https://json-schema.org/draft/2020-12/schema",
      "schemaID": "example-schema",
      "title": "Пример схемы",
      "description": "string",
      "type": "object",
      "properties": {
        "type": {
          "type": "string",
          "description": "Возможные типы объектов",
          "enum": [
            "File"
          ]
        },
        "keys": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "type": {
                        "type": "string"
                    },
                    "value": {
                        "type": "string"
                    }
                }
            }
        },
        "context": {
          "type": "object",
          "description": "Контекст объекта",
          "additionalProperties": false,
          "required": [
            "size"
          ],
          "properties": {
            "size": {
              "type": "integer",
              "minimum": 0,
              "description": "Размер файла в байтах"
            }
          }
        }
      }
    }"""

    schema_id = None
    schema = json.loads(jsonSchema)
    try:  # register object schema
        sch_ref = client.iocean.schemas.register(schema)
        schema_id = sch_ref.schema_id
    except ConflictError:
        # handle Duplicate Error here
        exit(1)

    view = client.iocean.schemas.view(schema_id)
    print(view.schema)

    client.close()

examples/register_object_schema.py
cyberstudio/cybsi-cloud-sdk