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