Source code for cybsi.cloud.client

from .auth import APIKeyAuth, AuthAPI
from .client_config import Config
from .files import FilesAPI, FilesAsyncAPI
from .insight.api import InsightAPI, InsightAsyncAPI
from .internal import AsyncHTTPConnector, HTTPConnector
from .iocean import IOCeanAPI, IOCeanAsyncAPI


[docs] 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, retry=config.retry, ) 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)
[docs] 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) @property def insight(self) -> InsightAPI: """Insight API handle.""" return InsightAPI(self._connector) @property def files(self) -> FilesAPI: """Files API handle.""" return FilesAPI(self._connector)
[docs] 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, retry=config.retry, ) 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)
[docs] 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) @property def insight(self) -> InsightAsyncAPI: """Insight asynchronous API handle.""" return InsightAsyncAPI(self._connector) @property def files(self) -> FilesAsyncAPI: """Files API handle.""" return FilesAsyncAPI(self._connector)