Kontekst: Otwarte dane jako infrastruktura publiczna

Otwarte dane we włoskiej administracji publicznej to nie tylko zasada przejrzystości: to obowiązek regulacyjne (Dekret Legislacyjny 36/2006, zmieniony Dekretem Legislacyjnym 200/2021 w ramach transpozycji Dyrektywy UE o Otwartych Danych 2019/1024), motorem innowacji gospodarczych i kluczowym elementem ekosystemu Platforma cyfrowa Dane krajowe (PDND).

Portal krajowy data.gov.it, zarządzany przez AgID, gromadzi i indeksuje ponad 5000 zbiorów danych Włoska administracja publiczna. Wszystkie opublikowane dane — od rozkładów jazdy komunikacji miejskiej po wyniki wybory, od danych o jakości powietrza po uchwały gmin — muszą spełniać standardy Precyzyjne metadatowanie, jakość i dostępność.

Dla programisty wyzwanie związane z otwartymi danymi rządowymi jest dwojakie: z jednej strony publikować dane tak, aby naprawdę nadawały się do ponownego wykorzystania (surowy plik CSV na stronie instytucji nie wystarczy), z drugiej strony konsumować heterogeniczne dane z różnych źródeł o zmiennych standardach i jakości. W tym artykule omówiono oba wymiary.

Czego się nauczysz

  • Standard DCAT-AP_IT: struktura katalogu, zbiór danych, dystrybucja i obowiązkowe metadane
  • CKAN: konfiguracja, włoskie rozszerzenia i REST API do zarządzania zbiorami danych
  • Projekt API REST dla otwartych danych: paginacja, filtrowanie, wersjonowanie i buforowanie
  • Formaty wdrożeniowe: CSV, JSON-LD, RDF, GeoJSON, Parquet
  • Jakość danych: walidacja, profilowanie i wskaźniki jakości DCAT
  • Otwarte zużycie danych: solidni klienci, obsługa błędów i normalizacja
  • PDND i interoperacyjność: publikuj i korzystaj z interfejsów API PA za pośrednictwem platformy krajowej

Standard DCAT-AP_IT: Metadatowanie danych publicznych

Włoski profil DCAT-AP (Profil aplikacji słownictwa katalogu danych), znany jako DCAT-AP_IT, to złoty standard publikowania metadanych zbiorów danych we włoskich PA. Jest zdefiniowany przez AgID i opiera się na specyfikacjach W3C DCAT, z określonymi rozszerzeniami dla kontekstu włoskiego (geografie, licencje, tematy EUROVOC itp.).

Główna struktura DCAT-AP_IT obejmuje trzy podstawowe podmioty:

  • Katalog (dcat:Catalog): katalog zbiorów danych PA. Zawiera metadane dotyczące organizacji który publikuje dane, domyślną licencję, stronę główną i okres aktualizacji.
  • Zbiory danych (dcat:Datasets): źródło informacji. Zawiera tytuł, opis, motywy i częstotliwość data aktualizacji, data utworzenia/modyfikacji, autor, czas i odniesienie geograficzne oraz konkretna licencja.
  • Dystrybucja (dcat:Dystrybucja): określony format, w jakim dostępny jest zbiór danych (CSV, JSON, RDF, plik kształtu itp.). Zawiera adres URL pobierania, format, rozmiar i datę ostatniej modyfikacji.
# Generatore di metadati DCAT-AP_IT in Python
# Produce RDF/Turtle compatibile con dati.gov.it

from rdflib import Graph, Literal, URIRef, Namespace
from rdflib.namespace import DCAT, DCT, FOAF, RDF, XSD
from datetime import datetime, date

# Namespace italiani
DCATAPIT = Namespace("http://dati.gov.it/onto/dcatapit#")
VCARD = Namespace("http://www.w3.org/2006/vcard/ns#")
SKOS = Namespace("http://www.w3.org/2004/02/skos/core#")

def create_dcat_ap_it_metadata(
    catalog_uri: str,
    dataset_id: str,
    title_it: str,
    description_it: str,
    publisher_name: str,
    publisher_uri: str,
    themes: list,
    license_uri: str,
    distributions: list
) -> str:
    """
    Genera metadati DCAT-AP_IT completi in formato Turtle.
    """
    g = Graph()
    g.bind("dcat", DCAT)
    g.bind("dct", DCT)
    g.bind("foaf", FOAF)
    g.bind("dcatapit", DCATAPIT)

    # Definisci il Dataset
    dataset_uri = URIRef(f"{catalog_uri}/dataset/{dataset_id}")

    g.add((dataset_uri, RDF.type, DCAT.Dataset))
    g.add((dataset_uri, RDF.type, DCATAPIT.Dataset))

    # Metadati obbligatori
    g.add((dataset_uri, DCT.title, Literal(title_it, lang="it")))
    g.add((dataset_uri, DCT.description, Literal(description_it, lang="it")))
    g.add((dataset_uri, DCT.modified, Literal(datetime.utcnow().date().isoformat(), datatype=XSD.date)))
    g.add((dataset_uri, DCT.accrualPeriodicity, URIRef("http://publications.europa.eu/resource/authority/frequency/MONTHLY")))
    g.add((dataset_uri, DCT.license, URIRef(license_uri)))

    # Publisher (obbligatorio in DCAT-AP_IT)
    publisher = URIRef(publisher_uri)
    g.add((publisher, RDF.type, DCATAPIT.Agent))
    g.add((publisher, FOAF.name, Literal(publisher_name, lang="it")))
    g.add((publisher, DCT.identifier, Literal(publisher_uri)))
    g.add((dataset_uri, DCT.publisher, publisher))

    # Temi dal vocabolario EU EUROVOC
    for theme_uri in themes:
        g.add((dataset_uri, DCAT.theme, URIRef(theme_uri)))

    # Distribuzione per ogni formato
    for i, dist in enumerate(distributions):
        dist_uri = URIRef(f"{dataset_uri}/distribution/{i}")
        g.add((dist_uri, RDF.type, DCAT.Distribution))
        g.add((dist_uri, RDF.type, DCATAPIT.Distribution))
        g.add((dist_uri, DCAT.accessURL, URIRef(dist["url"])))
        g.add((dist_uri, DCT.format, URIRef(f"http://publications.europa.eu/resource/authority/file-type/{dist['format']}")))
        g.add((dist_uri, DCT.license, URIRef(license_uri)))
        if "bytes" in dist:
            g.add((dist_uri, DCAT.byteSize, Literal(dist["bytes"], datatype=XSD.decimal)))
        g.add((dataset_uri, DCAT.distribution, dist_uri))

    return g.serialize(format="turtle")

# Esempio di utilizzo
rdf_metadata = create_dcat_ap_it_metadata(
    catalog_uri="https://dati.comune.milano.it",
    dataset_id="qualità-aria-2024",
    title_it="Qualità dell'Aria - Rilevazioni 2024",
    description_it="Dataset con le rilevazioni orarie dei sensori di qualità dell'aria nel Comune di Milano",
    publisher_name="Comune di Milano",
    publisher_uri="http://spcdata.digitpa.gov.it/browse/page/Amministrazione/agid",
    themes=["http://publications.europa.eu/resource/authority/data-theme/ENVI"],
    license_uri="https://creativecommons.org/licenses/by/4.0/",
    distributions=[
        {"url": "https://dati.comune.milano.it/dataset/aria-2024.csv", "format": "CSV"},
        {"url": "https://dati.comune.milano.it/dataset/aria-2024.json", "format": "JSON", "bytes": 45230000},
    ]
)

CKAN: portal otwartych danych włoskiej AP

CKAN (Comprehensive Knowledge Archive Network) to najbardziej rozpowszechniona platforma typu open source do zarządzania rządowymi portalami otwartych danych. Sama data.gov.it jest zbudowana na CKAN i tak samo z architektury korzystają dziesiątki włoskich gmin, regionów i ministerstw.

Rozszerzenie ckanext-dcatapit, opracowany przez GeoSolutions i Autonomiczną Prowincję Trydent, dodaje pełną obsługę profilu DCAT-AP_IT, umożliwiając CKAN udostępnianie i wykorzystywanie zgodnych metadanych zgodnie ze standardami włoskimi i europejskimi.

# Utilizzo delle API CKAN di dati.gov.it
# Le API CKAN sono REST con risposta JSON standardizzata

import httpx
import asyncio
from typing import Optional, List

class CKANClient:
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.headers = {"Content-Type": "application/json"}
        if api_key:
            self.headers["Authorization"] = api_key

    async def search_datasets(
        self,
        query: str,
        filters: Optional[dict] = None,
        rows: int = 20,
        start: int = 0
    ) -> dict:
        """
        Cerca dataset nel catalogo CKAN con filtraggio avanzato.
        L'API usa Solr internamente per la ricerca full-text.
        """
        params = {
            "q": query,
            "rows": rows,
            "start": start,
        }

        # Filtri Solr per raffinamento
        if filters:
            fq_parts = [f"{k}:{v}" for k, v in filters.items()]
            params["fq"] = " AND ".join(fq_parts)

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/api/3/action/package_search",
                params=params,
                headers=self.headers,
                timeout=30.0
            )
            response.raise_for_status()
            result = response.json()

            if not result.get("success"):
                raise ValueError(f"CKAN API error: {result.get('error')}")

            return {
                "total": result["result"]["count"],
                "datasets": result["result"]["results"],
                "page": start // rows + 1,
                "per_page": rows
            }

    async def get_dataset(self, dataset_id: str) -> dict:
        """Recupera un dataset specifico con tutte le sue distribuzioni."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/api/3/action/package_show",
                params={"id": dataset_id},
                headers=self.headers,
                timeout=30.0
            )
            response.raise_for_status()
            result = response.json()

            if not result.get("success"):
                raise ValueError(f"Dataset not found: {dataset_id}")

            return result["result"]

    async def create_dataset(self, dataset_metadata: dict) -> dict:
        """Pubblica un nuovo dataset (richiede API key con permessi di scrittura)."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/3/action/package_create",
                json=dataset_metadata,
                headers=self.headers,
                timeout=60.0
            )
            response.raise_for_status()
            return response.json()["result"]

# Utilizzo pratico
async def main():
    client = CKANClient("https://www.dati.gov.it")

    # Cerca dataset ambientali aggiornati
    results = await client.search_datasets(
        query="qualità aria",
        filters={"res_format": "CSV", "groups": "ambiente"},
        rows=10
    )

    print(f"Trovati {results['total']} dataset")
    for ds in results["datasets"]:
        print(f"- {ds['title']} ({ds['num_resources']} risorse)")

asyncio.run(main())

Projekt API REST dla Open Data

Kiedy agencja płatnicza publikuje swoje dane za pośrednictwem interfejsu API REST (a nie tylko za pośrednictwem CKAN), musi przestrzegać zasad projektowania które gwarantują użyteczność, stabilność i skalowalność. The Wytyczne dotyczące interoperacyjności Technika PA AgID definiuje wzorce REST, które należy stosować w przypadku usług publicznych.

Paginacja i filtrowanie

# FastAPI: REST API per open data con paginazione conforme AgID
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import JSONResponse
from typing import Optional, List
from datetime import date
import math

app = FastAPI(
    title="Open Data API - PA Example",
    description="API per la pubblicazione di dati aperti - conforme Linee Guida AgID",
    version="1.0.0"
)

@app.get("/api/v1/datasets/air-quality",
    summary="Rilevazioni qualità aria",
    tags=["Environmental Data"],
    response_model=dict)
async def get_air_quality(
    # Paginazione standard AgID: page + page_size
    page: int = Query(default=1, ge=1, description="Numero pagina (da 1)"),
    page_size: int = Query(default=100, ge=1, le=1000, description="Elementi per pagina (max 1000)"),
    # Filtraggio
    station_id: Optional[str] = Query(default=None, description="ID stazione di rilevamento"),
    pollutant: Optional[str] = Query(default=None, description="Inquinante (PM2.5, PM10, NO2, O3)"),
    date_from: Optional[date] = Query(default=None, description="Data inizio (ISO 8601)"),
    date_to: Optional[date] = Query(default=None, description="Data fine (ISO 8601)"),
    # Ordinamento
    sort_by: str = Query(default="timestamp", description="Campo di ordinamento"),
    sort_order: str = Query(default="desc", regex="^(asc|desc)$"),
    # Formato output
    format: str = Query(default="json", regex="^(json|csv|geojson)$")
):
    """
    Restituisce le rilevazioni di qualità dell'aria con paginazione e filtraggio.

    Supporta output in JSON, CSV e GeoJSON per compatibilità massima.
    Conforme a DCAT-AP_IT e Linee Guida interoperabilità AgID.
    """
    # Query al database con parametri
    offset = (page - 1) * page_size
    records, total_count = await air_quality_service.get_records(
        station_id=station_id,
        pollutant=pollutant,
        date_from=date_from,
        date_to=date_to,
        sort_by=sort_by,
        sort_order=sort_order,
        limit=page_size,
        offset=offset
    )

    total_pages = math.ceil(total_count / page_size)

    # Risposta con metadati di paginazione (pattern AgID)
    response_body = {
        "data": records,
        "meta": {
            "total_count": total_count,
            "page": page,
            "page_size": page_size,
            "total_pages": total_pages,
            "has_next": page < total_pages,
            "has_prev": page > 1,
        },
        "links": {
            "self": f"/api/v1/datasets/air-quality?page={page}&page_size={page_size}",
            "first": f"/api/v1/datasets/air-quality?page=1&page_size={page_size}",
            "last": f"/api/v1/datasets/air-quality?page={total_pages}&page_size={page_size}",
            "next": f"/api/v1/datasets/air-quality?page={page+1}&page_size={page_size}" if page < total_pages else None,
            "prev": f"/api/v1/datasets/air-quality?page={page-1}&page_size={page_size}" if page > 1 else None,
        },
        "dataset": {
            "id": "aria-qualità-2024",
            "title": "Qualità dell'Aria",
            "license": "CC BY 4.0",
            "publisher": "Comune di Milano",
            "last_updated": "2024-12-01"
        }
    }

    # Content negotiation: JSON vs CSV vs GeoJSON
    if format == "csv":
        return Response(
            content=records_to_csv(records),
            media_type="text/csv",
            headers={"Content-Disposition": "attachment; filename=aria-qualità.csv"}
        )
    elif format == "geojson":
        return Response(
            content=records_to_geojson(records),
            media_type="application/geo+json"
        )

    return JSONResponse(content=response_body)

Buforowanie i wydajność interfejsów API otwartych danych

Dane publiczne ze swej natury zmieniają się z przewidywalną częstotliwością (codziennie, co tydzień, co miesiąc). To czyni je idealnymi kandydatami do agresywnych strategii buforowania. Dobrze zaprojektowany interfejs API otwartych danych wykorzystuje nagłówki HTTP standard buforowania i może obsłużyć zdecydowaną większość żądań bez dotykania bazy danych.

# Strategia di caching per open data API con Redis
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
import redis.asyncio as redis
import json
import hashlib
from datetime import timedelta

class OpenDataCacheMiddleware:
    """
    Middleware di caching per API open data.
    Usa Redis come cache layer con TTL basato sulla frequenza di aggiornamento del dataset.
    """

    # TTL per tipo di dataset (in secondi)
    DATASET_TTL = {
        "realtime": 60,        # Dati real-time (qualità aria, traffico)
        "daily": 86400,        # Aggiornamento giornaliero
        "weekly": 604800,      # Aggiornamento settimanale
        "monthly": 2592000,    # Aggiornamento mensile
        "static": 31536000,    # Dati statici (confini amministrativi)
    }

    def __init__(self, app, redis_url: str):
        self.app = app
        self.redis = redis.from_url(redis_url)

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        request = Request(scope, receive)

        # Cache solo GET requests
        if request.method != "GET":
            await self.app(scope, receive, send)
            return

        # Genera cache key da URL + query params (ordinati per consistenza)
        cache_key = self._generate_cache_key(str(request.url))

        # Cerca nella cache
        cached = await self.redis.get(cache_key)
        if cached:
            response_data = json.loads(cached)
            response = Response(
                content=response_data["body"],
                status_code=response_data["status_code"],
                headers={
                    **response_data["headers"],
                    "X-Cache": "HIT",
                    "Cache-Control": "public, max-age=3600"
                }
            )
            await response(scope, receive, send)
            return

        # Esegui la request e intercetta la response
        response_body = []

        async def send_wrapper(message):
            if message["type"] == "http.response.body":
                response_body.append(message.get("body", b""))
            await send(message)

        await self.app(scope, receive, send_wrapper)

        # Salva in cache
        if response_body:
            body = b"".join(response_body)
            await self.redis.setex(
                cache_key,
                self.DATASET_TTL["daily"],  # Default: aggiornamento giornaliero
                json.dumps({"body": body.decode(), "status_code": 200, "headers": {}})
            )

    def _generate_cache_key(self, url: str) -> str:
        """Genera cache key stabile dall'URL."""
        return f"opendata:{hashlib.sha256(url.encode()).hexdigest()[:16]}"

Jakość danych: walidacja i profilowanie

Zbiór danych, który jest technicznie dostępny, ale zawiera dane niskiej jakości, nie jest naprawdę „otwarty” w tym sensie przydatne tego terminu. AgID określiła szczegółowe cechy w Trzyletnim Planie ICT na lata 2024-2026 wskaźniki jakości dla zbiorów danych PA, inspirowanych wymiarami jakości normy ISO/IEC 25012:

Wymiar jakości Definicja Praktyczna metryka Minimalny próg AgID
Kompletność Żadnych brakujących wartości % NULL pól z całości < 5% NULL w wymaganych polach
Dokładność Korespondencja z rzeczywistością Walidacja w oparciu o wiarygodne źródła To zależy od domeny
Konsystencja Wewnętrzna spójność zbioru danych Więzy referencyjne, kontrola zakresów 0% naruszeń ograniczeń
Aktualność Zbiór danych aktualizowany zgodnie z deklarowaną częstotliwością Dni od ostatniej aktualizacji a częstotliwość Aktualizacja w ciągu 2x podanego okresu
Zgodność Zgodność ze standardami (DCAT-AP_IT) Walidacja metadanych SHACL Obecność 100% obowiązkowych metadanych
# Data Quality Profiler per dataset PA
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class QualityReport:
    dataset_id: str
    total_rows: int
    total_columns: int
    quality_score: float  # 0-100
    issues: List[dict] = field(default_factory=list)
    column_stats: Dict[str, Any] = field(default_factory=dict)

class DataQualityProfiler:
    """
    Profiler di qualità per dataset open data PA.
    Calcola metriche ISO/IEC 25012 e produce un report strutturato.
    """

    REQUIRED_FIELDS = ["id", "timestamp", "value", "station_code"]

    def profile(self, df: pd.DataFrame, dataset_id: str) -> QualityReport:
        report = QualityReport(
            dataset_id=dataset_id,
            total_rows=len(df),
            total_columns=len(df.columns),
            quality_score=100.0
        )

        # 1. Completezza: campi obbligatori
        for field_name in self.REQUIRED_FIELDS:
            if field_name not in df.columns:
                report.issues.append({
                    "severity": "critical",
                    "dimension": "completeness",
                    "field": field_name,
                    "message": f"Campo obbligatorio '{field_name}' mancante"
                })
                report.quality_score -= 20
            else:
                null_pct = df[field_name].isna().sum() / len(df) * 100
                if null_pct > 5:
                    report.issues.append({
                        "severity": "warning",
                        "dimension": "completeness",
                        "field": field_name,
                        "message": f"{null_pct:.1f}% valori NULL nel campo '{field_name}'",
                        "null_count": int(df[field_name].isna().sum()),
                        "null_percentage": null_pct
                    })
                    report.quality_score -= min(10, null_pct)

        # 2. Consistenza: duplicati
        duplicate_count = df.duplicated().sum()
        if duplicate_count > 0:
            dup_pct = duplicate_count / len(df) * 100
            report.issues.append({
                "severity": "warning",
                "dimension": "consistency",
                "message": f"{duplicate_count} righe duplicate ({dup_pct:.1f}%)",
                "duplicate_count": int(duplicate_count)
            })
            report.quality_score -= min(15, dup_pct * 2)

        # 3. Statistiche per colonna
        for col in df.columns:
            col_stats = {
                "dtype": str(df[col].dtype),
                "null_count": int(df[col].isna().sum()),
                "null_percentage": df[col].isna().sum() / len(df) * 100,
                "unique_count": int(df[col].nunique()),
            }
            if df[col].dtype in [np.float64, np.int64]:
                col_stats.update({
                    "min": float(df[col].min()),
                    "max": float(df[col].max()),
                    "mean": float(df[col].mean()),
                    "median": float(df[col].median()),
                })
            report.column_stats[col] = col_stats

        report.quality_score = max(0.0, report.quality_score)
        return report

Korzystanie z otwartych danych: solidni klienci

Po stronie konsumpcji publiczne zbiory danych mają cechy wymagające szczególnej uwagi: mogą je mieć nieregularne aktualizacje, mogą być tymczasowo niedostępne, formaty mogą się różnić w zależności od wersji tego samego zbioru danych, a jakość danych nie jest gwarantowana. Solidny klient musi poradzić sobie ze wszystkimi te sytuacje.

# Client robusto per consumare open data PA
import httpx
import asyncio
from typing import AsyncIterator
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RobustOpenDataClient:
    """
    Client per consumo open data con retry, streaming e validazione.
    """

    def __init__(self, timeout: int = 60):
        self.timeout = timeout

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=60),
        retry=retry_if_exception_type((httpx.HTTPError, httpx.TimeoutException))
    )
    async def fetch_dataset(self, url: str) -> dict:
        """Scarica un dataset con retry automatico in caso di errore."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(url, follow_redirects=True)
            response.raise_for_status()
            return response.json()

    async def stream_large_csv(self, url: str) -> AsyncIterator[dict]:
        """
        Streama CSV di grandi dimensioni senza caricare tutto in memoria.
        Utile per dataset che possono essere di centinaia di MB.
        """
        import csv
        import io

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            async with client.stream("GET", url) as response:
                response.raise_for_status()

                buffer = ""
                headers = None

                async for chunk in response.aiter_text(chunk_size=8192):
                    buffer += chunk
                    lines = buffer.split("\n")

                    # Mantieni l'ultima linea incompleta nel buffer
                    buffer = lines[-1]
                    complete_lines = lines[:-1]

                    if headers is None and complete_lines:
                        headers = list(csv.reader([complete_lines[0]]))[0]
                        complete_lines = complete_lines[1:]

                    if headers:
                        for line in complete_lines:
                            if line.strip():
                                row = list(csv.reader([line]))[0]
                                if len(row) == len(headers):
                                    yield dict(zip(headers, row))

# Utilizzo: import dati ISTAT da API REST
async def import_istat_data():
    client = RobustOpenDataClient()

    # API ISTAT SDMX REST
    istat_url = "https://esploradati.istat.it/SDMXWS/rest/data/IT1,DCSC_POPRES1_EV,1.0/A.IT.9.0?startPeriod=2020"

    try:
        data = await client.fetch_dataset(istat_url)
        # Normalizza il formato SDMX
        return normalize_sdmx_response(data)
    except httpx.HTTPStatusError as e:
        raise RuntimeError(f"ISTAT API error {e.response.status_code}: {e.response.text}")

PDND: Krajowa Platforma Danych Cyfrowych

La Krajowa Platforma Danych Cyfrowych (PDND)zarządzana przez PagoPA S.p.A., to infrastruktura krajowy system interoperacyjności, który umożliwia agencjom płatniczym udostępnianie danych w sposób bezpieczny, kontrolowany i identyfikowalne. W przeciwieństwie do czystych otwartych danych (danych publicznych dostępnych dla każdego), PDND również zarządza dane wrażliwe, których udostępnianie jest dozwolone na mocy porozumień międzyinstytucjonalnych.

Dla programistów integracja PDND oznacza:

  • Dołącz do PDND jako użytkownik lub dostawca za pośrednictwem portalu interop.pagopa.it
  • Publikuj interfejsy API zgodnie z wytycznymi dotyczącymi interoperacyjności AgID (obowiązkowo OpenAPI 3.1, e-usługa z deskryptorem PDND)
  • Uwierzytelniać za pośrednictwem tokenów JWT podpisanych certyfikatami X.509 dla każdego żądania danych
  • Przestrzegaj kuponów użytkowych: umowy cyfrowe pomiędzy podmiotami autoryzującymi dostęp do określonych API

Wnioski i dalsze kroki

Wysokiej jakości otwarte dane rządowe wymagają czegoś więcej niż tylko opublikowania pliku CSV na stronie instytucji: wymaga standardowego metadatowania (DCAT-AP_IT), dobrze zaprojektowanego API REST z paginacją i buforowaniem, ciągłej walidacji jakość danych oraz integrację z infrastrukturą krajową, taką jak CKAN i PDND.

W następnym artykule z tej serii zajmiemy się dostępnym interfejsem użytkownika dla PA zgodnie z WCAG 2.1 AA: wymóg regulacyjny równie istotne, co wraz z otwartymi danymi przyczynia się do tego, aby publiczne usługi cyfrowe były naprawdę włączające.

Przydatne zasoby

Powiązane artykuły z tej serii

  • GovTech #00: Cyfrowa infrastruktura publiczna - elementy składowe i architektura
  • GovTech #04: RODO-by-Design - wzorce architektoniczne dla usług publicznych
  • GovTech #06: Integracja API rządowych - SPID, CIE i pagoPA
  • GovTech #07: GovStack Building Block - moduły dla cyfrowego rządu