gpf_entrepot_toolbelt.orchestrator.models.stored_data module

Modèle d’une donnée stockée sur l’Entrepôt de la GPF.

class gpf_entrepot_toolbelt.orchestrator.models.stored_data.GpfStoredData(_id: str, name: str, type: str, type_infos: dict = <factory>, status: str | None = None, srs: str | None = None, extent: ~gpf_entrepot_toolbelt.orchestrator.models.extent.GpfExtent | None = None, size: int | None = None, ancestors: list | None = None, storage: ~gpf_entrepot_toolbelt.orchestrator.models.storage.GpfStorage | None = None)

Bases : object

GPF StoredData class.

Exemple

output = GpfStoredData(
    _id="1231544456-1546546-164565",
    name="input_db",
    type="VECTOR-DB",
    type_infos={}
)
STORAGE_PREFIX = 'stored_data'
__init__(_id: str, name: str, type: str, type_infos: dict = <factory>, status: str | None = None, srs: str | None = None, extent: ~gpf_entrepot_toolbelt.orchestrator.models.extent.GpfExtent | None = None, size: int | None = None, ancestors: list | None = None, storage: ~gpf_entrepot_toolbelt.orchestrator.models.storage.GpfStorage | None = None) None
ancestors: list = None
check_type(allowed_stored_data_types: list[str]) Status

Function that checks if the stored data is one of the right types

Paramètres:

allowed_stored_data_types (list[str]) – list of allowed types.

Renvoie:

status of the check

Type renvoyé:

Status

property database: PgDatabase

Returns associated PgDatabase.

Raise ValueError in case of incompatible StoredData or missing informations.

Returns

PgDatabase: associated PgDatabase

extent: GpfExtent = None
classmethod from_dict(val: dict) Self

Load object from a dict.

Args

val (dict): dict value to load

Return

Self: object with attributes filled from dict.

Exemple

stored_data = GpfStoredData.from_dict(
        {
            "_id" : "1231544456-1546546-164565",
            "name" : "input_db",
            "type" : "VECTOR-DB",
            "type_infos" : {}
        }
    )
print(stored_data._id)
property fs_path: Path | None

Get the path to the filesystem, or None if stored_data is NOT on a FILESYSTEM storage type

Renvoie:

path to the filesystem

Type renvoyé:

Union[Path, None]

classmethod get_bbox_from_stored_datas(stored_datas: list[Self], srs: str = 'EPSG:4326') GpfBbox

Get global bounding box from Rok4 pyramid stored data list, in EPSG:4326

Paramètres:
  • stored_datas (list[GpfStoredData]) – pyramids list (stored datas)

  • srs (str, optional) – target projection

Renvoie:

global bounding box in EPSG:4326

Type renvoyé:

GpfBbox

property has_database: bool

Check if a stored data has an associated database

Returns

bool: True if a database is associated to stored data, False otherwise

import_files(dest: Path, strict: bool = True) tuple[list[str], list[str]]

Importe tous les fichiers définis de la stored_data dans le dossier spécifié.

Paramètres:
  • dest (Path) – dossier de destination des fichiers

  • strict (bool, optional) – active le mode strict. Defaults to True.

Lève:

S3IncompleteDownload – si le mode strict est activé et qu’il y a des téléchargements fichiers en erreur

Renvoie:

tuple des listes d’objets téléchargés avec succès et

ceux dont le téléchargement a échoué.

Type renvoyé:

tuple[str, str]

name: str
property readonly_database: PgDatabase

Returns associated PgDatabase with read only port if defined.

Raise ValueError in case of incompatible StoredData or missing informations.

Returns

PgDatabase: associated PgDatabase

size: int = None
srs: str = None
status: str = None
storage: GpfStorage = None
property storage_client: GpfS3Client | None

Get an authenticated storage client.

Renvoie:

object storage client

Type renvoyé:

Union[GpfS3Client, None]

property table_names_from_type_infos: list[str]

Returns list of table name from type_infos.

Returns

List[str]: list of table name

to_dict() dict

Convert dataclass object into dictionary. Supersedes dataclass.asdict

Renvoie:

object as dictionary

Type renvoyé:

dict

Exemple

output = GpfStoredData(
    _id="1231544456-1546546-164565",
    name="input_db",
    type="VECTOR-DB",
    type_infos={}
)
print(output.to_dict())
type: str
type_infos: dict