Source code for luna_licenses.crutches_on_wheels.cow.utils.functions

"""
Module realizes useful function.
"""
import asyncio
import contextlib
import os
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Callable, 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.utcnow() 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.utcnow(), 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 convertDateTimeToCurrentFormatStr(dateTimeString: str, currentTimeFormat: str) -> str: """ Convert datetime string from UTC or LOCAL to string, considering current STORAGE_TIME Args: dateTimeString: str with datetime currentTimeFormat: LOCAL or UTC Returns: str with datetime in current format """ dateTimeString = parse_datetime(dateTimeString).astimezone(UTC_TIMEZONE).replace(tzinfo=None) return convertTimeToString(dateTimeString, currentTimeFormat == "UTC")
[docs]def convertDateTimeToCurrentFormatDateTime(dateTimeString: str, currentTimeFormat: str) -> datetime: """ Convert datetime string from UTC or LOCAL to datetime Args: dateTimeString: str with datetime currentTimeFormat: LOCAL or UTC Returns: datetime with/without timezone depends on currentTimeFormat >>> t1 = convertDateTimeToCurrentFormatDateTime('2018-12-25 12:21:46.652495Z', 'LOCAL') >>> print(type(t1), t1) <class 'datetime.datetime'> 2018-12-25 12:21:46.652495 >>> t2 = convertDateTimeToCurrentFormatDateTime('2018-12-25 15:21:46.652495+03:00', 'LOCAL') >>> print(type(t2), t2) <class 'datetime.datetime'> 2018-12-25 12:21:46.652495 >>> t1 = convertDateTimeToCurrentFormatDateTime('2018-12-25 12:21:46.652495Z', 'UTC') >>> print(type(t1), t1) <class 'datetime.datetime'> 2018-12-25 12:21:46.652495+00:00 >>> t2 = convertDateTimeToCurrentFormatDateTime('2018-12-25 15:21:46.652495+03:00', 'UTC') >>> print(type(t2), t2) <class 'datetime.datetime'> 2018-12-25 15:21:46.652495+03:00 """ if currentTimeFormat == "UTC": return parse_datetime(dateTimeString) return parse_datetime(dateTimeString).astimezone(UTC_TIMEZONE).replace(tzinfo=None)
[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", "")