Bağlam: Kamu Altyapısı Olarak Açık Veri

İtalyan Kamu Yönetiminde açık veri yalnızca bir şeffaflık ilkesi değil aynı zamanda bir zorunluluktur düzenleyici (AB Açık Veri Direktifi 2019/1024'ün aktarılmasında 200/2021 sayılı Kanun Hükmünde Kararname ile değiştirilen 36/2006 sayılı Kanun Hükmünde Kararname), Ekonomik inovasyonun itici gücü ve ekosistemin önemli bir bileşeni Dijital Platform Ulusal Veriler (PDND).

Ulusal portal data.gov.itAgID tarafından yönetilen 5.000'den fazla veri kümesini toplar ve dizine ekler İtalyan kamu idareleri. Toplu taşıma programlarından sonuçlara kadar yayınlanan her veri Hava kalitesi verilerinden belediye kararlarına kadar seçimler, standartlara uygun olmalıdır. Hassas meta tarihleme, kalite ve erişilebilirlik.

Bir geliştirici için devletin açık verilerinin zorluğu iki yönlüdür: bir yandan yayınlamak veriler böylece gerçekten yeniden kullanılabilirler (kurumsal bir sitedeki ham CSV yeterli değildir), diğer yandan tüketmek Farklı kaynaklardan, değişken standartlara ve kaliteye sahip heterojen veriler. Bu makale her iki boyutu da ele almaktadır.

Ne Öğreneceksiniz

  • DCAT-AP_IT standardı: katalog yapısı, veri kümesi, dağıtım ve zorunlu meta veriler
  • CKAN: veri kümelerini yönetmek için yapılandırma, İtalyanca uzantılar ve REST API
  • Açık veriler için REST API tasarımı: sayfalandırma, filtreleme, sürüm oluşturma ve önbelleğe alma
  • Dağıtım formatları: CSV, JSON-LD, RDF, GeoJSON, Parquet
  • Veri kalitesi: doğrulama, profil oluşturma ve DCAT kalite ölçümleri
  • Açık veri tüketimi: sağlam istemciler, hata yönetimi ve normalleştirme
  • PDND ve birlikte çalışabilirlik: PA API'lerini ulusal platform aracılığıyla yayınlayın ve kullanın

DCAT-AP_IT Standardı: Genel Verilerin Meta Verilenmesi

İtalyan profili DCAT-AP (Veri Kataloğu Kelime Uygulama Profili), olarak bilinir DCAT-AP_IT, veri kümesi meta verilerini yayınlamak için altın standarttır İtalyan PA'larında. AgID tarafından tanımlanır ve belirli uzantılarla birlikte W3C DCAT spesifikasyonlarına dayanır. İtalya bağlamı için (coğrafyalar, lisanslar, EUROVOC temaları vb.).

DCAT-AP_IT'nin ana yapısı üç temel varlığı içerir:

  • Katalog (dcat:Katalog): Bir PA'nın veri kümelerinin kataloğu. Kuruluş hakkında meta veriler içerir verileri, varsayılan lisansı, ana sayfayı ve güncelleme dönemini yayınlayan.
  • Veri kümeleri (dcat:Veri kümeleri): bilgi kaynağı. Başlık, açıklama, temalar ve sıklığı içerir güncelleme tarihi, oluşturma/değiştirme tarihi, yazar, saat ve coğrafi referans ve özel lisans.
  • Dağıtım (dcat:Dağıtım): Veri kümesinin mevcut olduğu spesifik format (CSV, JSON, RDF, şekil dosyası vb.). İndirme URL'sini, biçimini, boyutunu ve son değiştirilme tarihini içerir. 
# 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: İtalyan PA'nın Açık Veri Portalı

CKAN (Kapsamlı Bilgi Arşivi Ağı) en yaygın açık kaynak platformudur Devlete ait açık veri portallarının yönetimi için. data.gov.it'in kendisi CKAN üzerine kuruludur ve aynı mimari onlarca İtalyan belediyesi, bölgesi ve bakanlığı tarafından kullanılıyor.

Uzantı ckanext-dcatapitGeoSolutions ve Trento Özerk Bölgesi tarafından geliştirilen, DCAT-AP_IT profili için tam destek ekleyerek CKAN'ın uyumlu meta verileri ortaya çıkarmasına ve kullanmasına olanak tanır İtalyan ve Avrupa standartlarına uygun. 

# 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())

Açık Veriler için REST API Tasarımı

Bir PA, verilerini REST API aracılığıyla (yalnızca CKAN aracılığıyla değil) yayınladığında tasarım ilkelerine uymalıdır Kullanılabilirliği, kararlılığı ve ölçeklenebilirliği garanti eden. Birlikte Çalışabilirlik Yönergeleri PA tekniği AgID, kamu hizmetleri için takip edilecek REST modellerini tanımlar.

Sayfalandırma ve Filtreleme

# 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)

Açık Veri API'leri için Önbelleğe Alma ve Performans

Kamuya açık veriler doğası gereği öngörülebilir bir sıklıkta (günlük, haftalık, aylık) değişmektedir. Bu onları agresif önbellekleme stratejileri için ideal adaylar haline getirir. İyi tasarlanmış bir açık veri API'si HTTP başlıklarını kullanır önbelleğe alma standardıdır ve isteklerin büyük çoğunluğunu veritabanına dokunmadan karşılayabilir. 

# 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]}"

Veri Kalitesi: Doğrulama ve Profil Oluşturma

Teknik olarak erişilebilir ancak kalitesiz verilere sahip bir veri kümesi, bu anlamda tam anlamıyla "açık" değildir. terimin faydası var. AgID, 2024-2026 Üç Yıllık BİT Planı'nda belirli özellikleri tanımladı kalite göstergeleri PA veri kümeleri için ISO/IEC 25012 standardının kalite boyutlarından esinlenilmiştir:

Kalite boyutu Tanım Pratik Metrik AgID Minimum Eşik
Tamlık Eksik değer yok Toplamdaki NULL alanların yüzdesi Zorunlu alanlarda < %5 NULL
Kesinlik Gerçeklikle uyumluluk Yetkili kaynaklara göre doğrulama Etki alanına bağlıdır
Tutarlılık Veri kümesinin iç tutarlılığı Referans kısıtlamaları, aralık kontrolleri %0 kısıtlama ihlali
Zamanındalık Veri seti beyan edilen sıklığa göre güncellendi Son güncellemeden bu yana geçen gün sayısı ve sıklık karşılaştırması Belirtilen sürenin 2 katı içinde güncellendi
Uyumluluk Standartlara uygunluk (DCAT-AP_IT) SHACL meta veri doğrulaması %100 zorunlu meta veri mevcut 
# 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

Açık Veri Tüketmek: Güçlü İstemciler

Tüketim tarafında, halka açık veri kümelerinin özel dikkat gerektiren özellikleri vardır: düzensiz güncellemeler, geçici olarak kullanılamayabilir, formatlar versiyonlar arasında farklılık gösterebilir aynı veri kümesinden oluşur ve verilerin kalitesi garanti edilmez. Güçlü bir müşteri bunların hepsini halletmelidir bu durumlar. 

# 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: Ulusal Dijital Veri Platformu

La Ulusal Dijital Veri Platformu (PDND)PagoPA S.p.A. tarafından yönetilen altyapıdır. KA'ların verileri güvenli, kontrollü ve güvenli bir şekilde paylaşmasına olanak tanıyan ulusal birlikte çalışabilirlik sistemi izlenebilir. Saf açık verilerden (herkesin erişebildiği kamuya açık veriler) farklı olarak PDND aynı zamanda paylaşımına kurumlar arası anlaşmalarla izin verilen hassas veriler.

Geliştiriciler için PDND'yi entegre etmek şu anlama gelir:

  • PDND'ye katılın interop.pagopa.it portalı aracılığıyla kullanıcı veya sağlayıcı olarak
  • API'leri yayınlayın AgID birlikte çalışabilirlik yönergelerini takip ederek (OpenAPI 3.1 zorunlu, PDND tanımlayıcılı e-hizmet)
  • Kimlik doğrulama her veri isteği için X.509 sertifikalarıyla imzalanmış JWT belirteçleri aracılığıyla
  • Kullanım kuponlarına saygı gösterin: belirli API'lere erişim yetkisi veren kuruluşlar arasındaki dijital anlaşmalar

Sonuçlar ve Sonraki Adımlar

Kaliteli devlet açık verileri, kurumsal bir sitede CSV dosyası yayınlamaktan daha fazlasını gerektirir: standart meta veri oluşturma (DCAT-AP_IT), sayfalandırma ve önbelleğe alma özelliğine sahip iyi tasarlanmış REST API, sürekli doğrulama gerektirir veri kalitesi ve CKAN ve PDND gibi ulusal altyapılarla entegrasyon.

Bu serinin bir sonraki makalesinde, WCAG 2.1 AA'ya göre PA için erişilebilir kullanıcı arayüzünü ele alacağız: düzenleyici bir gereklilik Aynı derecede kritik olan bu durum, açık verilerle birlikte kamuya ait dijital hizmetlerin gerçek anlamda kapsayıcı olmasına katkıda bulunur.

Yararlı Kaynaklar

Bu Serideki İlgili Makaleler

  • Devlet Teknolojisi #00: Dijital Kamu Altyapısı - yapı taşları ve mimarisi
  • Devlet Teknolojisi #04: GDPR-by-Design - kamu hizmetlerine yönelik mimari modeller
  • Devlet Teknolojisi #06: Devlet API entegrasyonu - SPID, CIE ve pagoPA
  • Devlet Teknolojisi #07: GovStack Building Block - dijital hükümet için modüller