オープンデータ API 設計: パブリック データの公開と利用
行政オープンデータの API を設計および実装する方法: DCAT-AP_IT、CKAN、標準 ページネーションとキャッシュを備えた REST API、オープン データ ポータル、データセット品質のベスト プラクティス。
背景: 公共インフラとしてのオープンデータ
イタリア行政におけるオープンデータは単なる透明性の原則ではなく、義務です 規制 (EU オープンデータ指令 2019/1024 を置き換えた立法令 200/2021 によって修正された立法令 36/2006)、 経済革新の原動力であり、エコシステムの重要な構成要素である デジタルプラットフォーム 国家データ (PDND).
全国ポータル データ政府IT、AgID によって管理され、5,000 を超えるデータセットを収集してインデックスを作成します イタリアの行政。公共交通機関のスケジュールから結果まで、公開されているすべてのデータ 選挙は、大気質データから地方自治体の決議に至るまで、次の基準に準拠する必要があります。 正確なメタデーティング、品質、アクセシビリティ。
開発者にとって、政府のオープンデータには 2 つの課題があります。 公開 データ 本当に再利用可能にするため (機関サイトの生の CSV だけでは十分ではありません)、もう 1 つは、 消費する さまざまなソースからの異質なデータを、さまざまな基準と品質で処理します。 この記事では両方の側面について説明します。
何を学ぶか
- DCAT-AP_IT 標準: カタログ構造、データセット、配布、および必須メタデータ
- CKAN: データセットを管理するための構成、イタリア語の拡張機能、および REST API
- オープン データ用の REST API 設計: ページネーション、フィルタリング、バージョン管理、キャッシュ
- デプロイ形式: CSV、JSON-LD、RDF、GeoJSON、Parquet
- データ品質: 検証、プロファイリング、DCAT 品質メトリクス
- オープンデータの消費: 堅牢なクライアント、エラー処理と正規化
- PDND と相互運用性: 全国プラットフォーム経由で PA API を公開および利用する
DCAT-AP_IT 標準: 公開データのメタデータ化
イタリア人のプロフィール DCAT-AP (Data Catalog Vocabulary Application Profile)、として知られています。 DCAT-AP_IT、データセットのメタデータを公開するためのゴールドスタンダードです イタリアのPAで。 AgID によって定義され、W3C DCAT 仕様に基づいており、特定の拡張子が付いています。 イタリアのコンテキスト (地理、ライセンス、EUROVOC テーマなど) に対応します。
DCAT-AP_IT の主な構造には、次の 3 つの基本的なエンティティが含まれます。
- カタログ (dcat:カタログ): PA のデータセットのカタログ。組織に関するメタデータが含まれます データ、デフォルトのライセンス、ホームページ、更新期間を公開します。
- データセット (dcat:データセット): 情報リソース。タイトル、説明、テーマ、頻度が含まれます 更新日、作成/変更日、作成者、時間と地理的参照、および特定のライセンス。
- ディストリビューション (dcat:ディストリビューション): データセットが利用可能な特定の形式 (CSV、JSON、RDF、シェープファイルなど)。ダウンロード URL、形式、サイズ、最終変更日が含まれます。
# 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: イタリア PA のオープン データ ポータル
CKAN (Comprehensive Knowledge Archive Network) は最も普及しているオープンソース プラットフォームです 政府のオープンデータポータルの管理用。 data.gov.it 自体は CKAN に基づいて構築されており、同じ この建築はイタリアの数十の自治体、地域、省庁で使用されています。
拡張子 ckanext-dcatapitGeoSolutions とトレント自治州によって開発されました。 DCAT-AP_IT プロファイルの完全なサポートを追加し、CKAN が準拠したメタデータを公開および使用できるようにします。 イタリアおよびヨーロッパの基準に準拠しています。
# 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())オープンデータ向けREST APIの設計
PA が (CKAN 経由だけでなく) REST API 経由でデータを公開する場合、設計原則に従う必要があります。 使いやすさ、安定性、拡張性を保証します。の 相互運用性ガイドライン PAテクニック AgID では、パブリック サービスに従う REST パターンを定義します。
ページネーションとフィルタリング
# 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)オープンデータ API のキャッシュとパフォーマンス
公開データはその性質上、予測可能な頻度 (毎日、毎週、毎月) で変化します。これ 積極的なキャッシュ戦略の理想的な候補となります。適切に設計されたオープン データ API は HTTP ヘッダーを使用します キャッシュの標準であり、データベースに触れることなく大部分のリクエストに対応できます。
# 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]}"データ品質: 検証とプロファイリング
技術的にはアクセス可能でもデータの品質が低いデータセットは、ある意味では真の意味で「オープン」ではありません。 この用語は役に立ちます。 AgID は、2024 年から 2026 年の 3 か年 ICT 計画で特定の機能を定義しました 品質指標 PA データセット用。ISO/IEC 25012 標準の品質次元からインスピレーションを受けています。
| 品質の次元 | 意味 | 実用的な指標 | AgID 最小しきい値 |
|---|---|---|---|
| 完全 | 欠損値なし | 合計のうち % NULL フィールド | 必須フィールドに < 5% NULL |
| 正確さ | 現実との対応 | 信頼できる情報源に対する検証 | ドメインによって異なります |
| 一貫性 | データセットの内部一貫性 | 参照制約、範囲チェック | 制約違反は 0% |
| 適時性 | 宣言された頻度に従ってデータセットが更新される | 前回の更新からの日数と頻度 | 指定期間の 2 倍以内に更新 |
| コンプライアンス | 規格への準拠 (DCAT-AP_IT) | SHACLメタデータの検証 | 100% 必須のメタデータが存在する |
# 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オープンデータの利用: 堅牢なクライアント
利用面では、公開データセットには特に注意が必要な特性があります。 不定期な更新、一時的に利用できない場合があります、バージョンごとに形式が異なる場合があります 同じデータセットのものであり、データの品質は保証されません。堅牢なクライアントはそれらすべてを処理する必要があります これらの状況。
# 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: 国家デジタル データ プラットフォーム
La 国家デジタル データ プラットフォーム (PDND)PagoPA S.p.A. が管理するインフラストラクチャです。 PA が安全かつ管理された環境でデータを共有できる国家相互運用システム。 追跡可能。純粋なオープンデータ (誰でもアクセスできる公開データ) とは異なり、PDND は管理も行います。 機関間の協定によって共有が許可されている機密データ。
開発者にとって、PDND の統合は次のことを意味します。
- PDNDに参加する interop.pagopa.it ポータル経由のユーザーまたはプロバイダーとして
- APIの公開 AgID 相互運用性ガイドライン (OpenAPI 3.1 必須、 PDND 記述子を使用した e サービス)
- 認証する データ要求ごとに X.509 証明書で署名された JWT トークン経由
- 利用券を尊重する: 特定の API へのアクセスを許可するエンティティ間のデジタル契約
結論と次のステップ
質の高い政府オープン データには、単に CSV ファイルを機関サイトに投稿する以上のものが必要です。 標準メタデート (DCAT-AP_IT)、ページネーションとキャッシュを備えた適切に設計された REST API、継続的検証が必要 データ品質、CKAN や PDND などの国家インフラストラクチャとの統合。
このシリーズの次の記事では、WCAG 2.1 AA (規制要件) に従って PA のアクセシブルな UI について取り上げます。 これも同様に重要であり、オープン データと組み合わせることで、公共デジタル サービスを真に包括的なものにするのに貢献します。
役立つリソース
- データ政府IT - イタリアの全国オープンデータポータル
- DCAT-AP_IT ガイドライン - AgID メタデータ標準
- ckanext-dcatapit - DCAT-AP_IT の CKAN 拡張
- PDND ドキュメント - 相互運用性プラットフォーム
このシリーズの関連記事
- ガバテック #00: デジタル公共インフラ - ビルディングブロックとアーキテクチャ
- ガバメントテック #04: GDPR-by-Design - 公共サービスのアーキテクチャ パターン
- ガバメントテック #06: 政府 API の統合 - SPID、CIE、pagoPA
- ガバテック #07: GovStack Building Block - デジタル政府用モジュール