rs_server_common.utils package

Submodules

rs_server_common.utils.logging module

Logging utility.

class rs_server_common.utils.logging.CustomFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

Custom logging formatter with colored text. See: https://stackoverflow.com/a/56944256

format(record)

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

class rs_server_common.utils.logging.Logging

Bases: object

Logging utility.

lock: For code synchronization

level: Minimal log level to use for all new logging instances.

classmethod default(name='rspy')

Return a default Logger class instance.

Parameters:: name – Logger name. You can pass __name__ to use your current module name.

level = 10

lock = <unlocked _thread.lock object>

rs_server_common.utils.opentelemetry module

OpenTelemetry utility

rs_server_common.utils.opentelemetry.init_traces(app: FastAPI, service_name: str)

Init instrumentation of OpenTelemetry traces.

Parameters:

app (fastapi.FastAPI) – FastAPI application
service_name (str) – service name

rs_server_common.utils.utils module

This module is used to share common functions between apis endpoints

class rs_server_common.utils.utils.EoDAGDownloadHandler(db_handler: DownloadStatus, thread_started: Event, station: str, product_id: str, name: str, local: str | None, obs: str | None)

Bases: object

Dataclass to store arguments needed for eodag download.

db_handler

An instance used to access the database.

Type:: DownloadStatus

thread_started

Event to signal the start of the download thread.

Type:: threading.Event

station

Station identifier (needed only for CADIP).

Type:: str

product_id

Identifier of the product to be downloaded.

Type:: str

name

Filename of the file to be downloaded.

Type:: str

local

Local path where the product will be stored

Type:: str | None

obs

Path to the S3 storage where the file will be uploaded

Type:: str | None

db_handler: DownloadStatus

local: str | None

name: str

obs: str | None

product_id: str

station: str

thread_started: Event

rs_server_common.utils.utils.create_stac_collection(products: List[EOProduct], feature_template: dict, stac_mapper: dict) → dict: This function create a STAC feature for each EOProduct based on a given template

rs_server_common.utils.utils.eodag_download(argument: EoDAGDownloadHandler, db, init_provider: Callable[[str], Provider], **kwargs)

Start the eodag download process.

This function initiates the eodag download process using the provided arguments. It sets up the necessary configurations, starts the download thread, and updates the download status in the database based on the outcome of the download.

Parameters:: argument (EoDAGDownloadHandler) – An instance of EoDAGDownloadHandler containing the arguments used in the downloading process

Note

The local and obs parameters are optionals: - local (str | None): Local path where the product will be stored. If this

parameter is not given, the local path where the file is stored will be set to a temporary one

obs (str | None): Path to S3 storage where the file will be uploaded, after a successfull download from CADIP
server. If this parameter is not given, the file will not be uploaded to the s3 storage.

Returns:: None
Raises:: RuntimeError – If there is an issue connecting to the S3 storage during the download.

rs_server_common.utils.utils.extract_eo_product(eo_product: EOProduct, mapper: dict) → dict: This function is creating key:value pairs from an EOProduct properties

rs_server_common.utils.utils.is_valid_date_format(date: str) → bool

Check if a string adheres to the expected date format “YYYY-MM-DDTHH:MM:SS.sssZ”.

Parameters:: date (str) – The string to be validated for the specified date format.
Returns:: True if the input string adheres to the expected date format, otherwise False.
Return type:: bool

rs_server_common.utils.utils.odata_to_stac(feature_template: dict, odata_dict: dict, odata_stac_mapper: dict) → dict: This function is used to map odata values to a given STAC template

rs_server_common.utils.utils.sort_feature_collection(feature_collection: dict, sortby: str) → dict: This function sorts a STAC feature collection based on a given criteria

rs_server_common.utils.utils.update_db(db: Session, db_product: DownloadStatus, estatus: EDownloadStatus, status_fail_message=None)

Update the download status of a product in the database.

This function attempts to update the download status of a product in the database. It retries the update operation for a maximum of three times, waiting 1 second between attempts.

Parameters:

db (sqlalchemy.orm.Session) – The database session.
db_product (DownloadStatus) – The product whose status needs to be updated.
estatus (EDownloadStatus) – The new download status.
status_fail_message (Optional[str]) – An optional message associated with the failure status.

Returns:

None

Raises:

sqlalchemy.exc.OperationalError – If the database update operation fails after multiple attempts.

Example

>>> update_db(db_session, product_instance, EDownloadStatus.DONE)

Note

This function is designed to update the download status in the database.
It retries the update operation for a maximum of three times.
If the update fails, an exception is raised, indicating an issue with the database.

rs_server_common.utils.utils.validate_inputs_format(interval: str) → Tuple[datetime, datetime]

Validate the format of the input time interval.

This function checks whether the input interval has a valid format (start_date/stop_date) and whether the start and stop dates are in a valid ISO 8601 format.

Parameters:: interval (-) – The time interval to be validated, with the following format: “2024-01-01T00:00:00Z/2024-01-02T23:59:59Z”
Returns:: A tuple containing: - start_date (datetime): The start date of the interval. - stop_date (datetime): The stop date of the interval.
Return type:: Tuple[datetime, datetime]

Note

The input interval should be in the format “start_date/stop_date”

(e.g., “2022-01-01T00:00:00Z/2022-01-02T00:00:00Z”). - This function checks for missing start/stop and validates the ISO 8601 format of start and stop dates. - If there is an error, err_code and err_text provide information about the issue.

rs_server_common.utils.utils.write_search_products_to_db(db_handler_class: DownloadStatus, products: EOProduct) → None

Process a list of products by adding them to the database if not already present.

This function iterates over a list of products. For each product, it checks whether the product is already registered in the database. If the product is not in the database, it is added with its relevant details. The function collects a list of product IDs and names for further processing.

Parameters:

products (-) – A list of product objects to be processed.
db_handler_class (-) – The database handler class used for database operations.

Returns:

A list of tuples, each containing the ‘id’ and ‘Name’ properties of a product.

Return type:

List[Tuple]

Raises:

- sqlalchemy.exc.OperationalError – If there’s an issue connecting to the database.

Note

The function assumes that ‘products’ is a list of objects with a ‘properties’ attribute, which is a dictionary containing keys ‘id’, ‘Name’, and ‘startTimeFromAscendingNode’.
‘get_db’ is a context manager that provides a database session.
‘EDownloadStatus’ is an enumeration representing download status.

Module contents

Utility modules.