gpf_entrepot_toolbelt.orchestrator.models.output module¶

Output object with related sugar.

class gpf_entrepot_toolbelt.orchestrator.models.output.OutputDataStructure(executionId: str, status: Status = Status.SUCCESS, failures: Iterable[str | None] = (None,), trace: str | None = None)¶

Bases : object

Output data structure.

Exemple

output = OutputDataStructure(
    executionId="1231544456-1546546-164565",
    status=Status.FAILURE,
    failures=("data.shp", "sirene.csv"),
    trace="[USER] No SRS, no cry",
)

__init__(executionId: str, status: Status = Status.SUCCESS, failures: Iterable[str | None] = (None,), trace: str | None = None) → None¶

executionId: str¶

failures: Iterable[str | None] = (None,)¶

status: Status = 0¶

to_dict() → dict¶

Convert dataclass object into dictionary. Supersedes dataclass.asdict to handle enum value.

Renvoie:: object as dictionary
Type renvoyé:: dict

Exemple

output = OutputDataStructure(
    executionId="1231544456-1546546-164565",
)
print(output.to_dict())

to_json(status_as_name: bool = True, **kwargs) → str¶

Supersedes json.dumps using the dictionary returned by to_dict().

Paramètres:: status_as_name (bool, optional) – if True dumps the status as name. If not, as int. Defaults to True.
Renvoie:: object serialized as JSON string
Type renvoyé:: str

Exemple

from pathlib import Path

# create output file
out_file = Path("/tmp/output.json")
out_file.parent.mkdir(parents=True, exist_ok=True)

# write into the file passing extra parameters to json.dumps
with out_file.open("w", encoding="UTF8") as wf:
    wf.write(output.to_json(sort_keys=True))

trace: str = None¶