L'esportazione

Un'esportazione è un lavoro che gira dalla parte del servizio: Export lo rappresenta mentre è in corso, ExportStatus dice a che punto è, Progress quanto ne resta, e Corpus è l'archivio una volta scaricato. Export sta in piedi da solo, identificato dal suo token, e sopravvive al processo che l'ha avviato.

Il percorso completo, dai criteri all'archivio riaperto da disco, sta in esportare un atto intero.

Gli stati di un'esportazione

stateDiagram-v2
    [*] --> TO_CONFIRM: start_export()
    TO_CONFIRM --> WAITING
    WAITING --> PROCESSING
    PROCESSING --> CONFIRMED_WITH_DELAY: ci vuole piu' tempo
    CONFIRMED_WITH_DELAY --> PROCESSING
    PROCESSING --> COMPLETED: download()
    PROCESSING --> FAILED
    PROCESSING --> OVERLOADED
    COMPLETED --> [*]
    FAILED --> [*]
    OVERLOADED --> [*]

I tre stati in fondo concludono l'attesa di wait; gli altri la fanno tornare a interrogare il servizio.

Il formato dell'archivio

Un ZIP con una cartella per atto, e dentro un documento JSON per versione:

LEGGE_19900807_241/1990-08-18_090G0294_ORIGINALE_V0.json
LEGGE_19900807_241/1990-08-18_090G0294_VIGENZA_1990-12-20_V1.json
LEGGE_19900807_241/1990-08-18_090G0294_VIGENZA_1991-01-23_V2.json

Il nome porta la data di pubblicazione in Gazzetta, il codice redazionale, la data da cui la versione vale e il suo numero progressivo. Nessun campo del documento riporta quella data: se i nomi non dichiarano la versione, Corpus rifiuta l'archivio con UnexpectedResponseError invece di leggerli tutti come «originale».

Export

Export(
    token: str,
    trasporto: Trasporto,
    *,
    format: Format = JSON,
    stato: ExportStatus = TO_CONFIRM,
    sleep: Callable[[float], None] = sleep,
    clock: Callable[[], float] = monotonic,
)

Un'esportazione, dalla richiesta all'archivio prodotto.

Codice sorgente in src/normattiva/esporta.py

def __init__(
    self,
    token: str,
    trasporto: Trasporto,
    *,
    format: Format = Format.JSON,
    stato: ExportStatus = ExportStatus.TO_CONFIRM,
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    self._token = token
    self._formato = format
    self._trasporto = trasporto
    self._stato = stato
    self._avanzamento = Progress()
    self._posizione: str | None = None
    self._sleep = sleep
    self._clock = clock

token property

token: str

Il token con cui riprendere questa esportazione da un altro processo.

format property

format: Format

Il format in cui è stato richiesto l'archivio.

status property

status: ExportStatus

L'ultimo stato dichiarato dal servizio.

progress property

progress: Progress

L'ultimo avanzamento dichiarato dal servizio, quando lo dichiara.

from_token classmethod

from_token(
    token: str,
    trasporto: Trasporto,
    *,
    format: Format = JSON,
) -> Export

Riprende un'esportazione già avviata, a partire dal suo token.

Codice sorgente in src/normattiva/esporta.py

@classmethod
def from_token(
    cls, token: str, trasporto: Trasporto, *, format: Format = Format.JSON
) -> Export:
    """Riprende un'esportazione già avviata, a partire dal suo token."""
    esportazione = cls(token, trasporto, format=format, stato=ExportStatus.WAITING)
    esportazione.refresh()
    return esportazione

refresh

refresh() -> ExportStatus

Interroga il servizio una volta sullo stato dell'esportazione.

Codice sorgente in src/normattiva/esporta.py

def refresh(self) -> ExportStatus:
    """Interroga il servizio una volta sullo stato dell'esportazione."""
    risposta = self._trasporto.get(
        f"ricerca-asincrona/check-status/{self._token}", attesi=(200, 202, 303)
    )
    stato, posizione, avanzamento = _stato_da(risposta, self._posizione)
    self._stato, self._posizione, self._avanzamento = stato, posizione, avanzamento
    return stato

wait

wait(*, timeout: float = TIMEOUT) -> ExportStatus

Interroga il servizio finché l'archivio è pronto, o finché la scadenza è superata.

Se il servizio dichiara un possibile ritardo, la scadenza viene prorogata una sola volta: prorogarla a ogni dichiarazione toglierebbe ogni limite all'attesa.

Parametri:

Nome	Tipo	Descrizione	Predefinito
timeout	float	quanti secondi attendere prima di rinunciare.	TIMEOUT

Restituisce:

Tipo	Descrizione
ExportStatus	Lo stato in cui l'esportazione si è conclusa.

Solleva:

Tipo	Descrizione
ExportFailedError	il servizio l'ha dichiarata fallita, oppure l'attesa ha superato timeout.
OverloadedError	il servizio non è in grado di completarla adesso.

Codice sorgente in src/normattiva/esporta.py

def wait(self, *, timeout: float = TIMEOUT) -> ExportStatus:
    """Interroga il servizio finché l'archivio è pronto, o finché la scadenza è superata.

    Se il servizio dichiara un possibile ritardo, la scadenza viene
    prorogata una sola volta: prorogarla a ogni dichiarazione toglierebbe
    ogni limite all'attesa.

    Args:
        timeout: quanti secondi attendere prima di rinunciare.

    Returns:
        Lo stato in cui l'esportazione si è conclusa.

    Raises:
        ExportFailedError: il servizio l'ha dichiarata fallita,
            oppure l'attesa ha superato `timeout`.
        OverloadedError: il servizio non è in grado di completarla adesso.
    """
    limite = self._clock() + timeout
    prorogato = False
    while True:
        stato = self.refresh()
        if stato.done:
            return stato
        if stato is ExportStatus.CONFIRMED_WITH_DELAY and not prorogato:
            limite = self._clock() + timeout
            prorogato = True
        if self._clock() >= limite:
            raise ExportFailedError(
                f"l'esportazione non si è conclusa entro {timeout:.0f} secondi"
            )
        logger.debug(
            "esportazione %s: stato %s, %s", self._token, stato.name, self._avanzamento
        )
        self._sleep(ATTESA_FRA_CONTROLLI)

download

download() -> Corpus

Scarica l'archivio e legge gli atti che contiene.

Solo il format JSON viene convertito in modelli; gli altri formati si scaricano come file con save, perché la libreria non li interpreta.

Restituisce:

Tipo	Descrizione
Corpus	Gli atti che l'archivio contiene, e l'archivio stesso.

Solleva:

Tipo	Descrizione
InvalidArgumentError	il format non è JSON; usare save.
UnexpectedResponseError	l'archivio non è leggibile, o i nomi dei file non dichiarano più la vigenza.

Codice sorgente in src/normattiva/esporta.py

def download(self) -> Corpus:
    """Scarica l'archivio e legge gli atti che contiene.

    Solo il format JSON viene convertito in modelli; gli altri formati si
    scaricano come file con `save`, perché la libreria non li interpreta.

    Returns:
        Gli atti che l'archivio contiene, e l'archivio stesso.

    Raises:
        InvalidArgumentError: il format non è JSON; usare `save`.
        UnexpectedResponseError: l'archivio non è leggibile, o i nomi dei file
            non dichiarano più la vigenza.
    """
    _verifica_leggibile(self._formato, "save()")
    return Corpus.from_data(self._scarica())

save

save(path: str | Path) -> Path

Scarica l'archivio e lo scrive su disco, in qualunque format.

Codice sorgente in src/normattiva/esporta.py

def save(self, path: str | Path) -> Path:
    """Scarica l'archivio e lo scrive su disco, in qualunque format."""
    destinazione = Path(path)
    destinazione.write_bytes(self._scarica())
    return destinazione

AsyncExport

AsyncExport(
    token: str,
    trasporto: TrasportoAsync,
    *,
    format: Format = JSON,
    stato: ExportStatus = TO_CONFIRM,
    sleep: Callable[[float], Awaitable[None]] = sleep,
    clock: Callable[[], float] = monotonic,
)

La variante asincrona di Export.

Codice sorgente in src/normattiva/esporta.py

def __init__(
    self,
    token: str,
    trasporto: TrasportoAsync,
    *,
    format: Format = Format.JSON,
    stato: ExportStatus = ExportStatus.TO_CONFIRM,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    self._token = token
    self._formato = format
    self._trasporto = trasporto
    self._stato = stato
    self._avanzamento = Progress()
    self._posizione: str | None = None
    self._sleep = sleep
    self._clock = clock

token property

token: str

Il token con cui riprendere questa esportazione da un altro processo.

format property

format: Format

Il format in cui è stato richiesto l'archivio.

status property

status: ExportStatus

L'ultimo stato dichiarato dal servizio.

progress property

progress: Progress

L'ultimo avanzamento dichiarato dal servizio, quando lo dichiara.

from_token async classmethod

from_token(
    token: str,
    trasporto: TrasportoAsync,
    *,
    format: Format = JSON,
) -> AsyncExport

Riprende un'esportazione già avviata, a partire dal suo token.

Codice sorgente in src/normattiva/esporta.py

@classmethod
async def from_token(
    cls, token: str, trasporto: TrasportoAsync, *, format: Format = Format.JSON
) -> AsyncExport:
    """Riprende un'esportazione già avviata, a partire dal suo token."""
    esportazione = cls(token, trasporto, format=format, stato=ExportStatus.WAITING)
    await esportazione.refresh()
    return esportazione

refresh async

refresh() -> ExportStatus

Interroga il servizio una volta sullo stato dell'esportazione.

Codice sorgente in src/normattiva/esporta.py

async def refresh(self) -> ExportStatus:
    """Interroga il servizio una volta sullo stato dell'esportazione."""
    risposta = await self._trasporto.get(
        f"ricerca-asincrona/check-status/{self._token}", attesi=(200, 202, 303)
    )
    stato, posizione, avanzamento = _stato_da(risposta, self._posizione)
    self._stato, self._posizione, self._avanzamento = stato, posizione, avanzamento
    return stato

wait async

wait(*, timeout: float = TIMEOUT) -> ExportStatus

Interroga il servizio finché l'archivio è pronto, o finché la scadenza è superata.

Se il servizio dichiara un possibile ritardo, la scadenza viene prorogata una sola volta: prorogarla a ogni dichiarazione toglierebbe ogni limite all'attesa.

Parametri:

Nome	Tipo	Descrizione	Predefinito
timeout	float	quanti secondi attendere prima di rinunciare.	TIMEOUT

Restituisce:

Tipo	Descrizione
ExportStatus	Lo stato in cui l'esportazione si è conclusa.

Solleva:

Tipo	Descrizione
ExportFailedError	il servizio l'ha dichiarata fallita, oppure l'attesa ha superato timeout.
OverloadedError	il servizio non è in grado di completarla adesso.

Codice sorgente in src/normattiva/esporta.py

async def wait(self, *, timeout: float = TIMEOUT) -> ExportStatus:
    """Interroga il servizio finché l'archivio è pronto, o finché la scadenza è superata.

    Se il servizio dichiara un possibile ritardo, la scadenza viene
    prorogata una sola volta: prorogarla a ogni dichiarazione toglierebbe
    ogni limite all'attesa.

    Args:
        timeout: quanti secondi attendere prima di rinunciare.

    Returns:
        Lo stato in cui l'esportazione si è conclusa.

    Raises:
        ExportFailedError: il servizio l'ha dichiarata fallita,
            oppure l'attesa ha superato `timeout`.
        OverloadedError: il servizio non è in grado di completarla adesso.
    """
    limite = self._clock() + timeout
    prorogato = False
    while True:
        stato = await self.refresh()
        if stato.done:
            return stato
        if stato is ExportStatus.CONFIRMED_WITH_DELAY and not prorogato:
            limite = self._clock() + timeout
            prorogato = True
        if self._clock() >= limite:
            raise ExportFailedError(
                f"l'esportazione non si è conclusa entro {timeout:.0f} secondi"
            )
        logger.debug(
            "esportazione %s: stato %s, %s", self._token, stato.name, self._avanzamento
        )
        await self._sleep(ATTESA_FRA_CONTROLLI)

download async

download() -> Corpus

Scarica l'archivio e legge gli atti che contiene.

Solo il format JSON viene convertito in modelli; gli altri formati si scaricano come file con save, perché la libreria non li interpreta.

Restituisce:

Tipo	Descrizione
Corpus	Gli atti che l'archivio contiene, e l'archivio stesso.

Solleva:

Tipo	Descrizione
InvalidArgumentError	il format non è JSON; usare save.
UnexpectedResponseError	l'archivio non è leggibile, o i nomi dei file non dichiarano più la vigenza.

Codice sorgente in src/normattiva/esporta.py

async def download(self) -> Corpus:
    """Scarica l'archivio e legge gli atti che contiene.

    Solo il format JSON viene convertito in modelli; gli altri formati si
    scaricano come file con `save`, perché la libreria non li interpreta.

    Returns:
        Gli atti che l'archivio contiene, e l'archivio stesso.

    Raises:
        InvalidArgumentError: il format non è JSON; usare `save`.
        UnexpectedResponseError: l'archivio non è leggibile, o i nomi dei file
            non dichiarano più la vigenza.
    """
    _verifica_leggibile(self._formato, "save()")
    return Corpus.from_data(await self._scarica())

save async

save(path: str | Path) -> Path

Scarica l'archivio e lo scrive su disco, in qualunque format.

Codice sorgente in src/normattiva/esporta.py

async def save(self, path: str | Path) -> Path:
    """Scarica l'archivio e lo scrive su disco, in qualunque format."""
    destinazione = Path(path)
    destinazione.write_bytes(await self._scarica())
    return destinazione

ExportStatus

Bases: IntEnum

Stato di avanzamento di un'esportazione.

done property

done: bool

Indica se lo stato è terminale: interrogare di nuovo il servizio non lo cambierà.

Progress dataclass

Progress(
    percent: float | None = None,
    processed: int | None = None,
    total: int | None = None,
)

L'avanzamento che il servizio dichiara per un'esportazione.

La percentuale da sola non dice se il lavoro sta procedendo: processed e total sì, e sono l'unico modo per capire se un'esportazione lunga è ferma o solo lenta. Il servizio non li invia sempre.

Corpus dataclass

Corpus(atti: tuple[AttoStorico, ...], archive: bytes = b'')

Gli atti contenuti in un archivio esportato, insieme all'archivio stesso.

attribuzione property

attribuzione: str

La riga di attribuzione richiesta dalla licenza.

from_zip classmethod

from_zip(path: str | Path) -> Corpus

Riapre un'esportazione salvata in precedenza, senza accedere alla rete.

Parametri:

Nome	Tipo	Descrizione	Predefinito
path	str | Path	il file ZIP scritto in precedenza da save.	obbligatorio

Restituisce:

Tipo	Descrizione
Corpus	Gli atti che l'archivio contiene, e l'archivio stesso.

Solleva:

Tipo	Descrizione
UnexpectedResponseError	l'archivio non è leggibile, o non segue la convenzione di nomi da cui si legge la data di vigenza.

Codice sorgente in src/normattiva/esporta.py

@classmethod
def from_zip(cls, path: str | Path) -> Corpus:
    """Riapre un'esportazione salvata in precedenza, senza accedere alla rete.

    Args:
        path: il file ZIP scritto in precedenza da `save`.

    Returns:
        Gli atti che l'archivio contiene, e l'archivio stesso.

    Raises:
        UnexpectedResponseError: l'archivio non è leggibile, o non segue la
            convenzione di nomi da cui si legge la data di vigenza.
    """
    dati = Path(path).read_bytes()
    return cls(atti=_wire.leggi_corpus(dati), archive=dati)

from_data classmethod

from_data(dati: bytes) -> Corpus

Legge un archivio già presente in memoria.

Codice sorgente in src/normattiva/esporta.py

@classmethod
def from_data(cls, dati: bytes) -> Corpus:
    """Legge un archivio già presente in memoria."""
    return cls(atti=_wire.leggi_corpus(dati), archive=dati)

save

save(path: str | Path) -> Path

Scrive l'archivio su disco, per riaprirlo senza una nuova esportazione.

Codice sorgente in src/normattiva/esporta.py

def save(self, path: str | Path) -> Path:
    """Scrive l'archivio su disco, per riaprirlo senza una nuova esportazione."""
    destinazione = Path(path)
    destinazione.write_bytes(self.archive)
    return destinazione