"""
Module realizes useful function.
"""
import asyncio
import contextlib
import os
from datetime import UTC, datetime
from enum import Enum, EnumType
from pathlib import Path
from typing import Callable, Literal, Optional, Type, Union
import aiohttp
from ciso8601 import parse_datetime # pylint: disable-msg=no-name-in-module
from luna3.common.exceptions import LunaApiException
from luna3.common.requests import makeRequest
from vlutils.helpers import UTC_TIMEZONE, convertTimeToString, convertToSnakeCase
from yarl import URL
from ..errors.errors import Error
from ..errors.exception import VLException
from ..utils.regexps import IMAGE_STORE_REGEXP
from .log import Logger
[docs]
def currentTimestamp(logTime: str = "LOCAL", dbType: str = "postgres") -> Union[str, datetime]:
"""
Function get time (local or UTC, depends on logTime).
The function is needed to correctly fill in the fields in different databases
Args:
logTime: LOCAL or UTC, LOCAL for default
dbType: db type (oracle/postgres), postgres for default
Returns:
Timestamp if format '%Y-%m-%dT%H:%M:%S.%fZ'
"""
timestamp = datetime.now(UTC).replace(tzinfo=None) if logTime == "UTC" else datetime.now()
if dbType == "oracle":
return timestamp
return timestamp.isoformat("T")
[docs]
def currentDateTime(currentTimeFormat: str) -> str:
"""
Function get current date-time in ISO format.
Args:
currentTimeFormat: LOCAL or UTC
Returns:
date-time in rfc3339 format (e.g. 1997-07-16T19:20:30.45+01:00)
"""
return convertTimeToString(datetime.now(UTC).replace(tzinfo=None), inUTC=currentTimeFormat == "UTC")
[docs]
def getPageCount(objectCount: int, pageSize: int) -> int:
"""
Calculate count of pages with adjusted page size and object count
Args:
objectCount: count of objects
pageSize: page size
Returns:
count of pages
"""
return (objectCount + pageSize - 1) // pageSize
[docs]
async def addCallBack(callableFunction: Callable, sleepTime: int = 0) -> None:
"""
Function creates infinity loop, where call function
Args:
callableFunction: function to call
sleepTime: time in secods if callback is periodic, in seconds
"""
while True:
if sleepTime:
await asyncio.sleep(sleepTime)
await callableFunction()
[docs]
def convertEnumToSnakeDict(enum: Type[Enum]) -> dict:
"""
Convert enum with camelCase name to dict with snake_case keys
Args:
enum: enum
Returns:
dict
"""
return convertToSnakeCase({case.name: case.value for case in enum})
[docs]
@contextlib.contextmanager
def workingDirectory(path: Path):
"""
Changes working directory and returns to previous on exit.
Args:
path: temporary working directory
"""
previousWD = Path.cwd()
os.chdir(path)
try:
yield
finally:
os.chdir(previousWD)
[docs]
async def downloadImage(
url: Union[str, URL], logger: Logger, timeout: aiohttp.ClientTimeout, accountId: Optional[str] = None
) -> tuple[bytes, str]:
"""
Download image. Add Luna-Account-Id header for request. If request to lis, apply account id to request.
Args:
url: image url
logger: logger
timeout: client timeout
accountId: account id
Returns:
tuple with image bytes and its content type
Raises:
VLException(Error.FailDownloadImage.format(url), 400, isCriticalError=False) if failed to download image
"""
preparedUrl = url if isinstance(url, URL) else URL(url)
headers = {"Accept-Encoding": "identity"}
if accountId is not None:
headers["Luna-Account-Id"] = accountId
if IMAGE_STORE_REGEXP.match(preparedUrl.path):
preparedUrl = preparedUrl.update_query(f"account_id={accountId}")
try:
resp = await makeRequest(
url=preparedUrl,
method="GET", # pylint: disable=duplicate-code
totalTimeout=timeout.total,
connectTimeout=timeout.connect,
sockConnectTimeout=timeout.sock_connect,
sockReadTimeout=timeout.sock_read,
headers=headers,
asyncRequest=True,
raiseError=True,
)
except LunaApiException as exc:
logger.warning("Cannot download an image", exc_info=True)
raise VLException(Error.FailDownloadImage.format(url), 400, isCriticalError=False) from exc
return resp.body, resp.headers.get("Content-Type", "")
[docs]
def getAnnotationWithNullable(value: EnumType | dict | list | tuple) -> type[Literal]:
"""
Get python annotation with allowed nullable value
Notes:
annotation contains 2 parts:
- first part is `tuple(...) + None` - required for good validation error messages
- second part `| None` - required to allow None as value
Args:
value: incoming value
Returns:
annotation
"""
if isinstance(value, EnumType):
return Literal[tuple(e.value for e in value)] | None
return Literal[tuple(list(value))] | None