"""
Module realizes useful function.
"""
import asyncio
import contextlib
import os
from datetime import UTC, datetime
from enum import Enum, EnumType
from numbers import Real
from pathlib import Path
from typing import Callable, Literal, Mapping, Optional, Type, Union
import aiohttp
import ujson # pylint: disable-msg=no-name-in-module
from ciso8601 import parse_datetime
from luna3.common.aiohttp_exception import guessAiohttpError
from luna3.common.exceptions import LunaApiException
from luna3.common.luna_request import LunaRequest
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
[docs]
async def simpleMakeRequest(
url: URL,
method: str,
queryParams: Optional[dict] = None,
body: Optional[Union[str, bytes, dict, list, aiohttp.Payload]] = None,
headers: Optional[dict] = None,
totalTimeout: Optional[Real] = None,
connectTimeout: Optional[Real] = None,
sockConnectTimeout: Optional[Real] = None,
sockReadTimeout: Optional[Real] = None,
) -> tuple[int, bytes, Mapping]:
"""
Make simple request.
Args:
url: request URL
method: request method
body: request body
queryParams: query parameters
headers: headers for request
totalTimeout: request's processing timeout in seconds.
connectTimeout: connection timeout in seconds.
sockConnectTimeout: socket connection timeout in seconds.
sockReadTimeout: socket chunk read timeout in seconds.
Returns:
status code, body bytes, headers.
"""
timeout = aiohttp.ClientTimeout(
total=totalTimeout, connect=connectTimeout, sock_connect=sockConnectTimeout, sock_read=sockReadTimeout
)
request = LunaRequest(
url=url,
method=method,
body=body,
params=queryParams,
headers=headers,
timeout=timeout,
)
try:
async with aiohttp.ClientSession() as session:
async with session.request(**request) as response:
return response.status, await response.read(), response.headers
except (asyncio.TimeoutError, aiohttp.ClientError) as exc:
if isinstance(exc, asyncio.TimeoutError):
status = 504
else:
status = 500
body = ujson.dumps(guessAiohttpError(exc, request))
headers = {"Content-Type": "application/json; charset=utf-8"}
return status, body.encode(), headers