Source code for luna_lambda_tools.public.clients.lis

"""
Module contains luna-image-store client adapted for usage in lambda
"""

from typing import Awaitable

from luna3.common.luna_response import LunaResponse
from luna3.image_store.image_store import StoreApi


[docs] class LIS: """Luna-image-store client"""
[docs] def __init__(self, lisClient: StoreApi, accountId: str, bucket: str): self._accountId = accountId self._lisClient = lisClient self._bucket = bucket
[docs] def getAddress(self) -> str: """Get luna-image-store address""" return f"{self._lisClient.baseUri}/buckets/{self._bucket}"
[docs] def postImage( self, imageInBytes: bytes | bytearray, contentType: str = "image/jpeg", raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Post an image to bucket. Args: imageInBytes: byte's array (image) contentType: content-type of image or image/jpeg for default raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.postImage( imageInBytes=imageInBytes, bucketName=self._bucket, contentType=contentType, raiseError=raiseError, accountId=self._accountId, **kwargs, )
[docs] def putImage( self, imageInBytes: bytes | bytearray, imageId: str, contentType: str = "image/jpeg", raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Put an image to bucket. Args: imageInBytes: byte's array (image) imageId: image id contentType: content-type of image or image/jpeg for default raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.putImage( imageInBytes=imageInBytes, imageId=imageId, bucketName=self._bucket, contentType=contentType, raiseError=raiseError, accountId=self._accountId, **kwargs, )
[docs] def headImage( self, imageId: str, withMeta: bool = None, raiseError=True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Check image existence in bucket. Args: imageId: external image id withMeta: whether to retrieve user-defined image metadata raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.checkImage( imageId=imageId, withMeta=withMeta, accountId=self._accountId, bucketName=self._bucket, raiseError=raiseError, **kwargs, )
[docs] def getImage( self, imageId: str, withMeta: bool = None, raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Get image from bucket. Args: imageId: external image id withMeta: whether to retrieve user-defined image metadata raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.getImage( imageId=imageId, withMeta=withMeta, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def deleteImage(self, imageId: str, raiseError: bool = True, **kwargs) -> Awaitable[LunaResponse] | LunaResponse: """ Delete image from bucket. Args: imageId: image id to delete raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.deleteImage( imageId=imageId, bucketName=self._bucket, raiseError=raiseError, accountId=self._accountId, **kwargs, )
[docs] def deleteImages( self, imageIds: list[str], raiseError: bool = True, **kwargs ) -> Awaitable[LunaResponse] | LunaResponse: """ Delete images from bucket. Args: imageIds: images ids to delete raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.deleteImages( imageIds=imageIds, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def postObject( self, objectBody: str, contentType: str = "application/json", raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Post object to bucket. Args: objectBody: object, available: text, json contentType: content-type of object or application/json for default raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.postObject( objectBody=objectBody, contentType=contentType, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def putObject( self, objectBody: str, objectId: str, contentType: str | None = "application/json", raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Put object to bucket. Args: objectBody: object, available: text, json objectId: object id contentType: content-type of object or application/json for default raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.putObject( objectBody=objectBody, objectId=objectId, contentType=contentType, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def getObject( self, objectId: str, raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Get object from bucket. Args: objectId: object id raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.getObject( objectId=objectId, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def headObject( self, objectId: str, raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Check object existence in bucket. Args: objectId: object id raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.checkObject( objectId=objectId, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def deleteObject( self, objectId: str, raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Delete object from bucket. Args: objectId: object id raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.deleteObject( objectId=objectId, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )
[docs] def deleteObjects( self, objectIds: list[str], raiseError: bool = True, **kwargs, ) -> Awaitable[LunaResponse] | LunaResponse: """ Delete objects from bucket. Args: objectIds: objects ids to delete raiseError: whether to raise LunaApiException in case of failure Returns: class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*. In body of :class: `~.LunaResponse` will return json with version. """ return self._lisClient.deleteObjects( objectIds=objectIds, bucketName=self._bucket, accountId=self._accountId, raiseError=raiseError, **kwargs, )