Il client asincrono

AsyncNormattiva rispecchia Normattiva metodo per metodo e firma per firma: cambiano await e async for, e gli argomenti del costruttore sono gli stessi, salvo http_client, che qui vuole un httpx.AsyncClient, e sleep, che passa da asyncio.sleep.

Più corutine che condividono lo stesso client condividono anche la sua autolimitazione, quindi si mettono in fila da sole. Quando conviene usarlo lo spiega lavorare in asincrono.

AsyncNormattiva

AsyncNormattiva(
    *,
    user_agent: str | None = None,
    timeout: float = 30.0,
    retries: int = 2,
    requests_per_second: float = 2.0,
    base_url: str = PRODUZIONE,
    http_client: AsyncClient | None = None,
    sleep: Callable[[float], Awaitable[None]] = sleep,
    clock: Callable[[], float] = monotonic,
)

La variante asincrona del client.

Ogni metodo rispecchia il suo gemello su Normattiva, firma compresa; quelli che iterano restituiscono iteratori asincroni. Anche gli argomenti del costruttore sono gli stessi, con http_client che qui vuole un httpx.AsyncClient e sleep che passa da asyncio.sleep.

Le corutine che condividono un client condividono anche la sua autolimitazione, e quindi si mettono in fila da sole.

Ogni metodo che tocca la rete può sollevare ConnectionError se il servizio non è raggiungibile, UnexpectedResponseError se la risposta non ha la forma che la libreria sa leggere, e RequestBlockedError se lo strato di protezione respinge la forma della richiesta. Le sezioni Raises dei singoli metodi elencano solo le eccezioni specifiche di ciascuno.

Codice sorgente in src/normattiva/client.py

def __init__(
    self,
    *,
    user_agent: str | None = None,
    timeout: float = 30.0,
    retries: int = 2,
    requests_per_second: float = 2.0,
    base_url: str = PRODUZIONE,
    http_client: httpx.AsyncClient | None = None,
    sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> None:
    self._trasporto = TrasportoAsync(
        base_url=base_url,
        user_agent=user_agent,
        timeout=timeout,
        retries=retries,
        requests_per_second=requests_per_second,
        http_client=http_client,
        sleep=sleep,
        clock=clock,
    )
    self._sleep = sleep
    self._clock = clock
    self._dizionari: dict[str, tuple[Tipologica, ...]] = {}

base_url property

base_url: str

L'indirizzo base del servizio a cui questo client si rivolge.

closed property

closed: bool

Se questo client è stato chiuso.

close async

close() -> None

Rilascia il pool di connessioni, se è stato creato da questo client.

Un client HTTP passato dall'esterno resta aperto: chiuderlo spetta a chi lo ha creato.

Codice sorgente in src/normattiva/client.py

async def close(self) -> None:
    """Rilascia il pool di connessioni, se è stato creato da questo client.

    Un client HTTP passato dall'esterno resta aperto: chiuderlo spetta a chi
    lo ha creato.
    """
    await self._trasporto.close()

dettaglio async

dettaglio(
    atto: Urn | str | AttoTrovato,
    *,
    vigenza: Vigenza = None,
    se_troncato: SeTroncato = "segnala",
) -> DettaglioAtto

Legge il testo di un atto o di un suo articolo.

atto è un URN, oppure un AttoTrovato uscito da una ricerca. Per i dodici tipi di atto su trenta la cui forma URN non è verificata la libreria passa dalle coordinate di Gazzetta, che il servizio accetta allo stesso modo. Quel percorso però non supporta le date: chiedere una vigenza per un atto raggiungibile solo così solleva un errore, il parametro non viene ignorato in silenzio.

Senza vigenza il servizio restituisce il testo vigente oggi, e nella risposta nulla dichiara a quale data corrisponde: indicarla è l'unico modo di saperlo. Quando si indica una data, la finestra restituita viene verificata contro di essa.

Con se_troncato="solleva" un articolo che sembra tagliato solleva TruncationError invece di limitarsi a segnalarlo in una proprietà.

Parametri:

Nome	Tipo	Descrizione	Predefinito
atto	Urn | str | AttoTrovato	un URN, la sua forma testuale, oppure un AttoTrovato uscito da una ricerca. Il comma viene rimosso automaticamente, perché il servizio lo rifiuta in ingresso.	obbligatorio
vigenza	Vigenza	il giorno a cui leggere il testo, "originale" per la prima pubblicazione, None per il testo di oggi.	None
se_troncato	SeTroncato	"segnala" registra il sospetto nella proprietà possibile_troncamento, "solleva" lo trasforma in eccezione.	'segnala'

Restituisce:

Tipo	Descrizione
DettaglioAtto	Il testo richiesto, con la finestra di vigenza in cui è valido.

Esempi:

Il testo dell'articolo 19 della legge 241 com'era nel 2000, con la finestra in cui quel testo è stato in vigore::

atto = normattiva.dettaglio(
    "urn:nir:stato:legge:1990-08-07;241~art19",
    vigenza=date(2000, 1, 1),
)
print(atto.finestra)  # 1994-01-01 → 2005-03-07

Solleva:

Tipo	Descrizione
NotFoundError	nessun atto risponde a quelle coordinate.
AmbiguityError	l'URN corrisponde a più atti pubblicati; i candidati sono elencati nell'eccezione.
NotYetInForceError	l'articolo non esisteva alla data richiesta.
ValidityMismatchError	il servizio ha risposto con una versione che non copre la data richiesta.
TruncationError	solo con se_troncato="solleva", e solo se il testo sembra tagliato.
InvalidUrnError	l'URN è malformato, oppure la forma URN di quel tipo di atto non è verificata.
InvalidArgumentError	la vigenza è indicata due volte, oppure è chiesta per un atto raggiungibile solo dalle coordinate di Gazzetta.

Codice sorgente in src/normattiva/client.py

async def dettaglio(
    self,
    atto: Urn | str | AttoTrovato,
    *,
    vigenza: Vigenza = None,
    se_troncato: SeTroncato = "segnala",
) -> DettaglioAtto:
    """Legge il testo di un atto o di un suo articolo.

    `atto` è un URN, oppure un `AttoTrovato` uscito da una ricerca. Per i
    dodici tipi di atto su trenta la cui forma URN non è verificata la
    libreria passa dalle coordinate di Gazzetta, che il servizio accetta
    allo stesso modo. Quel percorso però non supporta le date: chiedere una
    `vigenza` per un atto raggiungibile solo così solleva un errore, il
    parametro non viene ignorato in silenzio.

    Senza `vigenza` il servizio restituisce il testo vigente oggi, e nella
    risposta nulla dichiara a quale data corrisponde: indicarla è l'unico
    modo di saperlo. Quando si indica una data, la finestra restituita
    viene verificata contro di essa.

    Con `se_troncato="solleva"` un articolo che sembra tagliato solleva
    `TruncationError` invece di limitarsi a segnalarlo in una
    proprietà.

    Args:
        atto: un URN, la sua forma testuale, oppure un `AttoTrovato`
            uscito da una ricerca. Il comma viene rimosso automaticamente,
            perché il servizio lo rifiuta in ingresso.
        vigenza: il giorno a cui leggere il testo, `"originale"` per la
            prima pubblicazione, `None` per il testo di oggi.
        se_troncato: `"segnala"` registra il sospetto nella proprietà
            `possibile_troncamento`, `"solleva"` lo trasforma in eccezione.

    Returns:
        Il testo richiesto, con la finestra di vigenza in cui è valido.

    Examples:
        Il testo dell'articolo 19 della legge 241 com'era nel 2000, con la
        finestra in cui quel testo è stato in vigore::

            atto = normattiva.dettaglio(
                "urn:nir:stato:legge:1990-08-07;241~art19",
                vigenza=date(2000, 1, 1),
            )
            print(atto.finestra)  # 1994-01-01 → 2005-03-07

    Raises:
        NotFoundError: nessun atto risponde a quelle coordinate.
        AmbiguityError: l'URN corrisponde a più atti pubblicati; i
            candidati sono elencati nell'eccezione.
        NotYetInForceError: l'articolo non esisteva alla data
            richiesta.
        ValidityMismatchError: il servizio ha risposto con una versione che
            non copre la data richiesta.
        TruncationError: solo con `se_troncato="solleva"`, e solo se
            il testo sembra tagliato.
        InvalidUrnError: l'URN è malformato, oppure la forma URN di quel
            tipo di atto non è verificata.
        InvalidArgumentError: la vigenza è indicata due volte, oppure è
            chiesta per un atto raggiungibile solo dalle coordinate di
            Gazzetta.
    """
    _verifica_se_troncato(se_troncato)
    if isinstance(atto, AttoTrovato) and not atto.ha_urn:
        _verifica_senza_vigenza(atto, vigenza)
        codice, giorno = _coordinate_di_gazzetta(atto)
        return await self.dettaglio_da_gazzetta(codice, giorno, se_troncato=se_troncato)
    urn = atto.urn if isinstance(atto, AttoTrovato) else atto
    risposta = await self._trasporto.post(
        "atto/dettaglio-atto-urn", {"urn": str(_urn_con_vigenza(urn, vigenza))}
    )
    return _controlla_dettaglio(_wire.leggi_dettaglio(risposta.json()), vigenza, se_troncato)

dettaglio_da_gazzetta async

dettaglio_da_gazzetta(
    codice_redazionale: str,
    data: date,
    *,
    se_troncato: SeTroncato = "segnala",
) -> DettaglioAtto

Legge un atto a partire dalle sue coordinate di Gazzetta.

È il percorso per gli atti la cui forma URN non è verificata: il codice redazionale e la data di Gazzetta arrivano da una ricerca, e il servizio risponde anche per un decreto-legge luogotenenziale del 1917. Questo percorso però non supporta la multivigenza: restituisce sempre il testo vigente oggi, e per una data serve un URN.

Parametri:

Nome	Tipo	Descrizione	Predefinito
codice_redazionale	str	l'identificativo di Gazzetta dell'atto, come arriva da AttoTrovato.gazzetta.codice_redazionale.	obbligatorio
data	date	la data di pubblicazione in Gazzetta.	obbligatorio
se_troncato	SeTroncato	come per dettaglio.	'segnala'

Restituisce:

Tipo	Descrizione
DettaglioAtto	Il testo dell'atto, sempre nella versione vigente oggi.

Solleva:

Tipo	Descrizione
NotFoundError	nessun atto per quelle coordinate di Gazzetta.
InvalidArgumentError	il codice redazionale è vuoto.

Codice sorgente in src/normattiva/client.py

async def dettaglio_da_gazzetta(
    self,
    codice_redazionale: str,
    data: date,
    *,
    se_troncato: SeTroncato = "segnala",
) -> DettaglioAtto:
    """Legge un atto a partire dalle sue coordinate di Gazzetta.

    È il percorso per gli atti la cui forma URN non è verificata: il codice
    redazionale e la data di Gazzetta arrivano da una ricerca, e il servizio
    risponde anche per un decreto-legge luogotenenziale del 1917. Questo
    percorso però non supporta la multivigenza: restituisce sempre il testo
    vigente oggi, e per una data serve un URN.

    Args:
        codice_redazionale: l'identificativo di Gazzetta dell'atto, come
            arriva da `AttoTrovato.gazzetta.codice_redazionale`.
        data: la data di pubblicazione in Gazzetta.
        se_troncato: come per `dettaglio`.

    Returns:
        Il testo dell'atto, sempre nella versione vigente oggi.

    Raises:
        NotFoundError: nessun atto per quelle coordinate di Gazzetta.
        InvalidArgumentError: il codice redazionale è vuoto.
    """
    _verifica_se_troncato(se_troncato)
    risposta = await self._trasporto.post(
        "atto/dettaglio-atto", _corpo_gazzetta(codice_redazionale, data)
    )
    return _controlla_dettaglio(_wire.leggi_dettaglio(risposta.json()), None, se_troncato)

cronologia async

cronologia(
    urn: Urn | str, *, massimo: int | None = None
) -> AsyncIterator[DettaglioAtto]

Percorre tutte le versioni di un articolo, dall'originale a quella in vigore.

Costa una richiesta per versione: le finestre di vigenza sono contigue, quindi ogni versione viene chiesta al giorno successivo alla fine della precedente. Con massimo l'iterazione si ferma dopo quel numero di versioni; senza, arriva all'ultima. Senza massimo la catena si ferma comunque dopo cinquecento passi: nessun articolo italiano ha cinquecento versioni, quindi una catena così lunga indica finestre non contigue.

Parametri:

Nome	Tipo	Descrizione	Predefinito
urn	Urn | str	l'articolo di cui percorrere le versioni.	obbligatorio
massimo	int | None	quante versioni al più produrre. Senza, si arriva in fondo.	None

Produce:

Tipo	Descrizione
AsyncIterator[DettaglioAtto]	Una versione per volta, dalla più vecchia alla più recente.

Solleva:

Tipo	Descrizione
UnexpectedResponseError	la catena non si chiude entro cinquecento passi, cioè le finestre di vigenza non sono contigue.

Codice sorgente in src/normattiva/client.py

async def cronologia(
    self, urn: Urn | str, *, massimo: int | None = None
) -> AsyncIterator[DettaglioAtto]:
    """Percorre tutte le versioni di un articolo, dall'originale a quella in vigore.

    Costa una richiesta per versione: le finestre di vigenza sono contigue,
    quindi ogni versione viene chiesta al giorno successivo alla fine della
    precedente. Con `massimo` l'iterazione si ferma dopo quel numero di
    versioni; senza, arriva all'ultima. Senza `massimo` la catena si ferma
    comunque dopo cinquecento passi: nessun articolo italiano ha
    cinquecento versioni, quindi una catena così lunga indica finestre non
    contigue.

    Args:
        urn: l'articolo di cui percorrere le versioni.
        massimo: quante versioni al più produrre. Senza, si arriva in fondo.

    Yields:
        Una versione per volta, dalla più vecchia alla più recente.

    Raises:
        UnexpectedResponseError: la catena non si chiude entro cinquecento
            passi, cioè le finestre di vigenza non sono contigue.
    """
    base = Urn.parse(urn).senza_comma
    prossima: Vigenza = "originale"
    prodotte = 0
    while massimo is None or prodotte < massimo:
        if massimo is None and prodotte >= PASSI_MASSIMI:
            raise UnexpectedResponseError(
                f"la catena delle versioni non si chiude dopo {PASSI_MASSIMI} passi: "
                "le finestre di vigenza non sono contigue"
            )
        atto = await self.dettaglio(base.con_vigenza(prossima))
        yield atto
        prodotte += 1
        chiusura = atto.finestra.fine if atto.finestra else None
        if chiusura is None:
            return
        prossima = chiusura + timedelta(days=1)

ricerca async

ricerca(
    testo: str,
    *,
    pagina: int = 1,
    per_pagina: int = 20,
    sort: Sort | str = NEWEST,
    tipo: str | None = None,
    anno: int | None = None,
    emettitore: str | None = None,
) -> EsitoRicerca

Cerca nel testo pieno del corpus.

Le parole vengono combinate in AND dal servizio; non c'è modo di chiedere un OR. tipo, anno ed emettitore sono le faccette che la risposta stessa propone, non le coordinate dell'atto: per quelle c'è ricerca_avanzata.

Parametri:

Nome	Tipo	Descrizione	Predefinito
testo	str	le parole da cercare, combinate in AND dal servizio.	obbligatorio
pagina	int	quale pagina di risultati, a partire da 1.	1
per_pagina	int	quanti risultati per pagina.	20
sort	Sort | str	"newest" dal più recente, "oldest" dal più vecchio.	NEWEST
tipo	str | None	codice della faccetta per tipo di atto, come "PLE".	None
anno	int | None	faccetta per anno di provvedimento.	None
emettitore	str | None	faccetta per amministrazione emanante.	None

Restituisce:

Tipo	Descrizione
EsitoRicerca	Una pagina di risultati, con il totale e le faccette per restringere.

Codice sorgente in src/normattiva/client.py

async def ricerca(
    self,
    testo: str,
    *,
    pagina: int = 1,
    per_pagina: int = 20,
    sort: Sort | str = Sort.NEWEST,
    tipo: str | None = None,
    anno: int | None = None,
    emettitore: str | None = None,
) -> EsitoRicerca:
    """Cerca nel testo pieno del corpus.

    Le parole vengono combinate in AND dal servizio; non c'è modo di
    chiedere un OR. `tipo`, `anno` ed `emettitore` sono le faccette che la
    risposta stessa propone, non le coordinate dell'atto: per quelle c'è
    `ricerca_avanzata`.

    Args:
        testo: le parole da cercare, combinate in AND dal servizio.
        pagina: quale pagina di risultati, a partire da 1.
        per_pagina: quanti risultati per pagina.
        sort: `"newest"` dal più recente, `"oldest"` dal più vecchio.
        tipo: codice della faccetta per tipo di atto, come `"PLE"`.
        anno: faccetta per anno di provvedimento.
        emettitore: faccetta per amministrazione emanante.

    Returns:
        Una pagina di risultati, con il totale e le faccette per restringere.
    """
    corpo = _corpo_ricerca(
        testo,
        pagina=pagina,
        per_pagina=per_pagina,
        sort=sort,
        tipo=tipo,
        anno=anno,
        emettitore=emettitore,
    )
    risposta = await self._trasporto.post("ricerca/semplice", corpo)
    return _wire.leggi_ricerca(risposta.json())

ricerca_avanzata async

ricerca_avanzata(
    *,
    denominazione: str | None = None,
    anno: int | None = None,
    numero: int | str | None = None,
    giorno: int | None = None,
    mese: int | None = None,
    titolo: str | None = None,
    testo: str | None = None,
    vigente_al: date | None = None,
    classe: ClasseProvvedimento | int | None = None,
    emanazione: Intervallo | None = None,
    pubblicazione: Intervallo | None = None,
    sort: Sort | str = NEWEST,
    tipo: str | None = None,
    emettitore: str | None = None,
    pagina: int = 1,
    per_pagina: int = 20,
) -> EsitoRicerca

Cerca per coordinate invece che per parole.

Senza nessuna coordinata il servizio risponde con l'intero corpus: è una richiesta ammessa, e la libreria non la rifiuta.

Parametri:

Nome	Tipo	Descrizione	Predefinito
denominazione	str | None	il tipo di atto come lo scrive il dizionario, per esempio "LEGGE" o "DECRETO LEGISLATIVO".	None
anno	int | None	anno di emanazione.	None
numero	int | str | None	numero del provvedimento.	None
giorno	int | None	giorno di emanazione.	None
mese	int | None	mese di emanazione.	None
titolo	str | None	parole da cercare nel titolo.	None
testo	str | None	parole da cercare nel testo.	None
vigente_al	date | None	tiene solo gli atti in vigore in quel giorno.	None
classe	ClasseProvvedimento | int | None	la classe redazionale dell'atto (senza aggiornamenti, aggiornato, abrogato).	None
emanazione	Intervallo | None	intervallo di emanazione, come coppia (dal, al). Un estremo può essere None per lasciare la finestra aperta.	None
pubblicazione	Intervallo | None	intervallo di pubblicazione in Gazzetta, come sopra.	None
sort	Sort | str	"newest" dal più recente, "oldest" dal più vecchio.	NEWEST
tipo	str | None	faccetta per tipo di atto.	None
emettitore	str | None	faccetta per amministrazione emanante.	None
pagina	int	quale pagina di risultati, a partire da 1.	1
per_pagina	int	quanti risultati per pagina.	20

Restituisce:

Tipo	Descrizione
EsitoRicerca	Una pagina di risultati, nella stessa forma che rende ricerca.

Codice sorgente in src/normattiva/client.py

async def ricerca_avanzata(
    self,
    *,
    denominazione: str | None = None,
    anno: int | None = None,
    numero: int | str | None = None,
    giorno: int | None = None,
    mese: int | None = None,
    titolo: str | None = None,
    testo: str | None = None,
    vigente_al: date | None = None,
    classe: ClasseProvvedimento | int | None = None,
    emanazione: Intervallo | None = None,
    pubblicazione: Intervallo | None = None,
    sort: Sort | str = Sort.NEWEST,
    tipo: str | None = None,
    emettitore: str | None = None,
    pagina: int = 1,
    per_pagina: int = 20,
) -> EsitoRicerca:
    """Cerca per coordinate invece che per parole.

    Senza nessuna coordinata il servizio risponde con l'intero corpus: è
    una richiesta ammessa, e la libreria non la rifiuta.

    Args:
        denominazione: il tipo di atto come lo scrive il dizionario, per
            esempio `"LEGGE"` o `"DECRETO LEGISLATIVO"`.
        anno: anno di emanazione.
        numero: numero del provvedimento.
        giorno: giorno di emanazione.
        mese: mese di emanazione.
        titolo: parole da cercare nel titolo.
        testo: parole da cercare nel testo.
        vigente_al: tiene solo gli atti in vigore in quel giorno.
        classe: la classe redazionale dell'atto (senza aggiornamenti,
            aggiornato, abrogato).
        emanazione: intervallo di emanazione, come coppia `(dal, al)`. Un
            estremo può essere `None` per lasciare la finestra aperta.
        pubblicazione: intervallo di pubblicazione in Gazzetta, come sopra.
        sort: `"newest"` dal più recente, `"oldest"` dal più vecchio.
        tipo: faccetta per tipo di atto.
        emettitore: faccetta per amministrazione emanante.
        pagina: quale pagina di risultati, a partire da 1.
        per_pagina: quanti risultati per pagina.

    Returns:
        Una pagina di risultati, nella stessa forma che rende `ricerca`.
    """
    corpo = _corpo_avanzata(
        _coordinate(
            denominazione,
            anno,
            numero,
            giorno,
            mese,
            titolo,
            testo,
            vigente_al,
            classe,
            emanazione,
            pubblicazione,
        ),
        sort,
        pagina,
        per_pagina,
        tipo,
        emettitore,
    )
    risposta = await self._trasporto.post("ricerca/avanzata", corpo)
    return _wire.leggi_ricerca(risposta.json())

ricerca_completa async

ricerca_completa(
    testo: str,
    *,
    massimo: int | None = None,
    per_pagina: int = 50,
    sort: Sort | str = NEWEST,
    tipo: str | None = None,
    anno: int | None = None,
    emettitore: str | None = None,
) -> AsyncIterator[AttoTrovato]

Scorre tutti gli atti che una ricerca trova, una pagina per volta.

L'iteratore è pigro: ogni pagina viene chiesta solo quando serve, quindi consumare dieci risultati costa una richiesta sola. Con massimo ci si ferma dopo quel numero di atti; per sapere quanti ce n'erano in tutto basta una ricerca e il suo totale.

Quel totale è anche la condizione di uscita: il numero di pagina restituito dal servizio non è affidabile, e usarlo come condizione potrebbe rileggere la prima pagina all'infinito.

Parametri:

Nome	Tipo	Descrizione	Predefinito
testo	str	le parole da cercare.	obbligatorio
massimo	int | None	quanti atti al più produrre. Senza, si arriva in fondo.	None
per_pagina	int	quanti risultati chiedere per richiesta.	50
sort	Sort | str	"newest" dal più recente, "oldest" dal più vecchio.	NEWEST
tipo	str | None	faccetta per tipo di atto.	None
anno	int | None	faccetta per anno di provvedimento.	None
emettitore	str | None	faccetta per amministrazione emanante.	None

Produce:

Tipo	Descrizione
AsyncIterator[AttoTrovato]	Un atto per volta, nell'ordine in cui il servizio li rende.

Codice sorgente in src/normattiva/client.py

async def ricerca_completa(
    self,
    testo: str,
    *,
    massimo: int | None = None,
    per_pagina: int = 50,
    sort: Sort | str = Sort.NEWEST,
    tipo: str | None = None,
    anno: int | None = None,
    emettitore: str | None = None,
) -> AsyncIterator[AttoTrovato]:
    """Scorre tutti gli atti che una ricerca trova, una pagina per volta.

    L'iteratore è pigro: ogni pagina viene chiesta solo quando serve,
    quindi consumare dieci risultati costa una richiesta sola. Con
    `massimo` ci si ferma dopo quel numero di atti; per sapere quanti ce
    n'erano in tutto basta una `ricerca` e il suo `totale`.

    Quel `totale` è anche la condizione di uscita: il numero di pagina
    restituito dal servizio non è affidabile, e usarlo come condizione
    potrebbe rileggere la prima pagina all'infinito.

    Args:
        testo: le parole da cercare.
        massimo: quanti atti al più produrre. Senza, si arriva in fondo.
        per_pagina: quanti risultati chiedere per richiesta.
        sort: `"newest"` dal più recente, `"oldest"` dal più vecchio.
        tipo: faccetta per tipo di atto.
        anno: faccetta per anno di provvedimento.
        emettitore: faccetta per amministrazione emanante.

    Yields:
        Un atto per volta, nell'ordine in cui il servizio li rende.
    """
    if massimo is not None and massimo <= 0:
        return
    prodotti = 0
    pagina = 1
    dichiarati: int | None = None
    quanti_per_pagina = per_pagina if massimo is None else min(per_pagina, massimo)
    while True:
        esito = await self.ricerca(
            testo,
            pagina=pagina,
            per_pagina=quanti_per_pagina,
            sort=sort,
            tipo=tipo,
            anno=anno,
            emettitore=emettitore,
        )
        if dichiarati is None:
            dichiarati = esito.totale
        for atto in esito.atti:
            yield atto
            prodotti += 1
            if massimo is not None and prodotti >= massimo:
                return
        if not esito.atti or prodotti >= dichiarati or pagina >= PAGINE_MASSIME:
            return
        pagina += 1

atti_aggiornati async

atti_aggiornati(
    dal: date, al: date
) -> AsyncIterator[AttoTrovato]

Elenca gli atti modificati fra due date.

Il flusso contiene solo le modifiche: un atto pubblicato dentro la finestra ma mai modificato dopo non compare. Le finestre più lunghe di un anno vengono spezzate, perché il servizio le rifiuta.

Parametri:

Nome	Tipo	Descrizione	Predefinito
dal	date	primo giorno della finestra, compreso.	obbligatorio
al	date	ultimo giorno della finestra, compreso.	obbligatorio

Produce:

Tipo	Descrizione
AsyncIterator[AttoTrovato]	Un atto modificato per volta, finestra dopo finestra.

Solleva:

Tipo	Descrizione
RuleViolationError	al precede dal. Sollevata prima di toccare la rete, col codice RuleCode.DATE_INVERTITE.

Codice sorgente in src/normattiva/client.py

async def atti_aggiornati(self, dal: date, al: date) -> AsyncIterator[AttoTrovato]:
    """Elenca gli atti modificati fra due date.

    Il flusso contiene solo le modifiche: un atto pubblicato dentro la
    finestra ma mai modificato dopo non compare. Le finestre più lunghe di
    un anno vengono spezzate, perché il servizio le rifiuta.

    Args:
        dal: primo giorno della finestra, compreso.
        al: ultimo giorno della finestra, compreso.

    Yields:
        Un atto modificato per volta, finestra dopo finestra.

    Raises:
        RuleViolationError: `al` precede `dal`. Sollevata prima di toccare
            la rete, col codice `RuleCode.DATE_INVERTITE`.
    """
    _verifica_intervallo(dal, al)
    for inizio, fine in _intervalli(dal, al):
        risposta = await self._trasporto.post(
            "ricerca/aggiornati", _corpo_aggiornati(inizio, fine)
        )
        for atto in _wire.leggi_ricerca(risposta.json()).atti:
            yield atto

denominazioni async

denominazioni(
    *, reload: bool = False
) -> tuple[Tipologica, ...]

Elenca i tipi di atto che il corpus contiene, e tiene il risultato in memoria.

Codice sorgente in src/normattiva/client.py

async def denominazioni(self, *, reload: bool = False) -> tuple[Tipologica, ...]:
    """Elenca i tipi di atto che il corpus contiene, e tiene il risultato in memoria."""
    return await self._dizionario(
        "denominazioni", "tipologiche/denominazione-atto", _wire.leggi_denominazioni, reload
    )

classi_provvedimento async

classi_provvedimento(
    *, reload: bool = False
) -> tuple[Tipologica, ...]

Elenca le classi redazionali a cui un atto può appartenere.

Codice sorgente in src/normattiva/client.py

async def classi_provvedimento(self, *, reload: bool = False) -> tuple[Tipologica, ...]:
    """Elenca le classi redazionali a cui un atto può appartenere."""
    return await self._dizionario(
        "classi", "tipologiche/classe-provvedimento", _wire.leggi_classi, reload
    )

export_formats async

export_formats(
    *, reload: bool = False
) -> tuple[Tipologica, ...]

Elenca i formati in cui si può chiedere un'esportazione.

Codice sorgente in src/normattiva/client.py

async def export_formats(self, *, reload: bool = False) -> tuple[Tipologica, ...]:
    """Elenca i formati in cui si può chiedere un'esportazione."""
    return await self._dizionario(
        "estensioni", "tipologiche/estensioni", _wire.leggi_estensioni, reload
    )

ricerche_predefinite async

ricerche_predefinite() -> tuple[RicercaPredefinita, ...]

Elenca le ricerche predefinite che il servizio propone.

Codice sorgente in src/normattiva/client.py

async def ricerche_predefinite(self) -> tuple[RicercaPredefinita, ...]:
    """Elenca le ricerche predefinite che il servizio propone."""
    risposta = await self._trasporto.get("ricerca/predefinita")
    return _wire.leggi_ricerche_predefinite(risposta.json())

collections async

collections() -> tuple[Collection, ...]

Elenca gli archivi già confezionati che il servizio mette a disposizione.

Codice sorgente in src/normattiva/client.py

async def collections(self) -> tuple[Collection, ...]:
    """Elenca gli archivi già confezionati che il servizio mette a disposizione."""
    risposta = await self._trasporto.get("collections/collection-predefinite")
    return _wire.leggi_collezioni(risposta.json())

download_collection async

download_collection(
    name: str,
    *,
    format: Format | str = JSON,
    mode: ExportMode | str = VIGENTE,
) -> Corpus

Scarica un archivio già confezionato e legge gli atti che contiene.

Solleva:

Tipo	Descrizione
InvalidArgumentError	il formato chiesto non viene letto in modelli; usare save_collection per averlo come file.

Codice sorgente in src/normattiva/client.py

async def download_collection(
    self,
    name: str,
    *,
    format: Format | str = Format.JSON,
    mode: ExportMode | str = ExportMode.VIGENTE,
) -> Corpus:
    """Scarica un archivio già confezionato e legge gli atti che contiene.

    Raises:
        InvalidArgumentError: il formato chiesto non viene letto in
            modelli; usare `save_collection` per averlo come file.
    """
    _verifica_leggibile(Format(format), "save_collection()")
    return Corpus.from_data(await self._archivio_collezione(name, format, mode))

save_collection async

save_collection(
    name: str,
    path: str | Path,
    *,
    format: Format | str = JSON,
    mode: ExportMode | str = VIGENTE,
) -> Path

Scarica su disco un archivio già confezionato, in qualunque formato sia.

Codice sorgente in src/normattiva/client.py

async def save_collection(
    self,
    name: str,
    path: str | Path,
    *,
    format: Format | str = Format.JSON,
    mode: ExportMode | str = ExportMode.VIGENTE,
) -> Path:
    """Scarica su disco un archivio già confezionato, in qualunque formato sia."""
    destinazione = Path(path)
    destinazione.write_bytes(await self._archivio_collezione(name, format, mode))
    return destinazione

start_export async

start_export(
    *,
    format: Format | str = JSON,
    mode: ExportMode | str = MULTIVIGENTE,
    massimo_atti: int | None = 100,
    escludi_testo: str | None = None,
    escludi_titolo: str | None = None,
    denominazione: str | None = None,
    anno: int | None = None,
    numero: int | str | None = None,
    giorno: int | None = None,
    mese: int | None = None,
    titolo: str | None = None,
    testo: str | None = None,
    vigente_al: date | None = None,
    classe: ClasseProvvedimento | int | None = None,
    emanazione: Intervallo | None = None,
    pubblicazione: Intervallo | None = None,
) -> AsyncExport

Chiede l'esportazione completa degli atti che una ricerca trova.

Le coordinate sono le stesse di ricerca_avanzata. Prima di avviare l'esportazione gli atti vengono contati con una ricerca sincrona, perché un'esportazione costa minuti di lavoro al servizio e, una volta partita, non si annulla. Con massimo_atti=None si parte senza conteggio.

escludi_testo ed escludi_titolo escludono dal risultato gli atti che contengono quelle parole, e sono l'unico filtro che l'esportazione supporta e la ricerca no: il conteggio preventivo non ne tiene conto, quindi può contare più atti di quanti ne arriveranno.

Parametri:

Nome	Tipo	Descrizione	Predefinito
format	Format | str	in che formato produrre l'archivio. Solo JSON viene poi letto in modelli.	JSON
mode	ExportMode | str	quante versioni includere nell'archivio.	MULTIVIGENTE
massimo_atti	int | None	il tetto oltre il quale l'esportazione non parte. None la avvia senza conteggio preventivo.	100
escludi_testo	str | None	esclude gli atti che contengono questa parola.	None
escludi_titolo	str | None	esclude gli atti il cui titolo la contiene.	None
denominazione	str | None	come in ricerca_avanzata, e così tutti i criteri che seguono.	None
anno	int | None	anno di emanazione.	None
numero	int | str | None	numero del provvedimento.	None
giorno	int | None	giorno di emanazione.	None
mese	int | None	mese di emanazione.	None
titolo	str | None	parole da cercare nel titolo.	None
testo	str | None	parole da cercare nel testo.	None
vigente_al	date | None	tiene solo gli atti in vigore in quel giorno.	None
classe	ClasseProvvedimento | int | None	la classe redazionale dell'atto (senza aggiornamenti, aggiornato, abrogato).	None
emanazione	Intervallo | None	intervallo di emanazione, come coppia (dal, al).	None
pubblicazione	Intervallo | None	intervallo di pubblicazione in Gazzetta.	None

Restituisce:

Tipo	Descrizione
AsyncExport	L'esportazione appena avviata, da attendere e poi scaricare.

Solleva:

Tipo	Descrizione
TooManyResultsError	i criteri selezionano più atti di massimo_atti. L'esportazione non viene avviata.
ConnectionError	il conteggio preventivo non è riuscito. Il messaggio indica come procedere senza.

Codice sorgente in src/normattiva/client.py

async def start_export(
    self,
    *,
    format: Format | str = Format.JSON,
    mode: ExportMode | str = ExportMode.MULTIVIGENTE,
    massimo_atti: int | None = 100,
    escludi_testo: str | None = None,
    escludi_titolo: str | None = None,
    denominazione: str | None = None,
    anno: int | None = None,
    numero: int | str | None = None,
    giorno: int | None = None,
    mese: int | None = None,
    titolo: str | None = None,
    testo: str | None = None,
    vigente_al: date | None = None,
    classe: ClasseProvvedimento | int | None = None,
    emanazione: Intervallo | None = None,
    pubblicazione: Intervallo | None = None,
) -> AsyncExport:
    """Chiede l'esportazione completa degli atti che una ricerca trova.

    Le coordinate sono le stesse di `ricerca_avanzata`. Prima di avviare
    l'esportazione gli atti vengono contati con una ricerca sincrona,
    perché un'esportazione costa minuti di lavoro al servizio e, una volta
    partita, non si annulla. Con `massimo_atti=None` si parte senza
    conteggio.

    `escludi_testo` ed `escludi_titolo` escludono dal risultato gli atti
    che contengono quelle parole, e sono l'unico filtro che l'esportazione
    supporta e la ricerca no: il conteggio preventivo non ne tiene conto,
    quindi può contare più atti di quanti ne arriveranno.

    Args:
        format: in che formato produrre l'archivio. Solo `JSON` viene poi
            letto in modelli.
        mode: quante versioni includere nell'archivio.
        massimo_atti: il tetto oltre il quale l'esportazione non parte.
            `None` la avvia senza conteggio preventivo.
        escludi_testo: esclude gli atti che contengono questa parola.
        escludi_titolo: esclude gli atti il cui titolo la contiene.
        denominazione: come in `ricerca_avanzata`, e così tutti i criteri
            che seguono.
        anno: anno di emanazione.
        numero: numero del provvedimento.
        giorno: giorno di emanazione.
        mese: mese di emanazione.
        titolo: parole da cercare nel titolo.
        testo: parole da cercare nel testo.
        vigente_al: tiene solo gli atti in vigore in quel giorno.
        classe: la classe redazionale dell'atto (senza aggiornamenti,
            aggiornato, abrogato).
        emanazione: intervallo di emanazione, come coppia `(dal, al)`.
        pubblicazione: intervallo di pubblicazione in Gazzetta.

    Returns:
        L'esportazione appena avviata, da attendere e poi scaricare.

    Raises:
        TooManyResultsError: i criteri selezionano più atti di
            `massimo_atti`. L'esportazione non viene avviata.
        ConnectionError: il conteggio preventivo non è riuscito. Il
            messaggio indica come procedere senza.
    """
    coordinate = _coordinate(
        denominazione,
        anno,
        numero,
        giorno,
        mese,
        titolo,
        testo,
        vigente_al,
        classe,
        emanazione,
        pubblicazione,
    )
    corpo = _corpo_export(
        coordinate,
        format,
        mode,
        {"testoNot": escludi_testo, "titoloNot": escludi_titolo},
    )
    if massimo_atti is not None:
        try:
            esito = await self.ricerca_avanzata(**coordinate, per_pagina=1)
        except (ConnectionError, UnexpectedResponseError) as errore:
            raise _conteggio_non_riuscito(errore) from errore
        if esito.totale > massimo_atti:
            raise TooManyResultsError(esito.totale, massimo_atti)
    risposta = await self._trasporto.post(
        "ricerca-asincrona/nuova-ricerca", corpo, attesi=(200, 202)
    )
    token = risposta.testo
    await self._trasporto.put(
        "ricerca-asincrona/conferma-ricerca", {"token": token}, attesi=(200, 202, 204)
    )
    return AsyncExport(
        token,
        self._trasporto,
        format=Format(format),
        sleep=self._sleep,
        clock=self._clock,
    )

export_from_token async

export_from_token(
    token: str, *, format: Format | str = JSON
) -> AsyncExport

Riprende un'esportazione già in corso, dal suo token.

Codice sorgente in src/normattiva/client.py

async def export_from_token(
    self, token: str, *, format: Format | str = Format.JSON
) -> AsyncExport:
    """Riprende un'esportazione già in corso, dal suo token."""
    return await AsyncExport.from_token(token, self._trasporto, format=Format(format))