本文目录导读:

构建Python数据API的数据目录,本质上是在数据源和API端点之间建立一个结构化的映射关系,这不仅是一个技术问题,更是一个组织与设计问题。
一个良好的数据目录(Data Catalog)通常包含元数据(字段类型、描述、来源)、访问路径(端点、查询参数)和数据本身的引用。
以下分几个层面,结合代码示例,说明如何构建。
核心思路:三层抽象
- 定义层 (Schema Definition):定义数据集的“蓝图”,包括名称、描述、字段、类型。
- 注册层 (Registry/Catalog):一个中心化的注册表(通常是字典或数据库表),管理所有可用的数据集。
- 访问层 (Data Retrieval):根据目录条目,实际加载并返回数据。
轻量级纯内存目录(适合小项目、微服务)
适用于数据量不大、启动时加载的情况,使用Python字典或dataclass。
定义元数据结构
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
import pandas as pd
@dataclass
class DatasetSchema:
"""单个数据集的元数据定义"""
id: str # 唯一标识,如 "sales.orders"
name: str # 人类可读名称
description: str
source_type: str # "csv", "parquet", "sql", "api"
source_path: str # 文件路径、表名或API URL
fields: List[Dict[str, Any]] = field(default_factory=list) # [{"name": "id", "type": "int", "desc": "..."}]
tags: List[str] = field(default_factory=list)
load_function: Optional[Callable] = None # 自定义加载函数
构建注册表 (Catalog Registry)
class DataCatalog:
"""简单的内存数据目录"""
def __init__(self):
self._datasets: Dict[str, DatasetSchema] = {}
self._data_cache: Dict[str, Any] = {}
def register(self, schema: DatasetSchema):
"""注册一个数据集定义"""
if schema.id in self._datasets:
raise ValueError(f"Dataset '{schema.id}' already registered.")
self._datasets[schema.id] = schema
def list_datasets(self) -> List[DatasetSchema]:
"""列出所有可用数据集"""
return list(self._datasets.values())
def get_schema(self, dataset_id: str) -> Optional[DatasetSchema]:
"""获取数据集的元数据"""
return self._datasets.get(dataset_id)
def load_data(self, dataset_id: str) -> Any:
"""根据目录定义加载实际数据(可缓存)"""
if dataset_id in self._data_cache:
return self._data_cache[dataset_id]
schema = self._datasets.get(dataset_id)
if not schema:
raise KeyError(f"Dataset '{dataset_id}' not found in catalog.")
# 根据 source_type 执行加载逻辑
if schema.load_function:
data = schema.load_function(schema.source_path)
elif schema.source_type == "csv":
data = pd.read_csv(schema.source_path)
elif schema.source_type == "parquet":
data = pd.read_parquet(schema.source_path)
else:
raise NotImplementedError(f"Unsupported source type: {schema.source_type}")
# 可选:进行字段类型校验
self._data_cache[dataset_id] = data
return data
集成到FastAPI中(示例)
from fastapi import FastAPI, HTTPException, Query
from typing import Optional
app = FastAPI()
catalog = DataCatalog()
# 初始化时注册数据源
catalog.register(DatasetSchema(
id="finance.revenue",
name="Revenue Data",
description="Monthly revenue by region",
source_type="csv",
source_path="data/revenue.csv",
fields=[
{"name": "month", "type": "date", "desc": "月份"},
{"name": "region", "type": "string", "desc": "地区"},
{"name": "amount", "type": "float", "desc": "收入金额"}
],
tags=["finance", "revenue"]
))
# 端点:列出所有可用的数据集(API的数据目录)
@app.get("/catalog")
def list_catalog():
return [{"id": ds.id, "name": ds.name, "description": ds.description, "tags": ds.tags}
for ds in catalog.list_datasets()]
# 端点:获取某个数据集的元数据
@app.get("/catalog/{dataset_id}/schema")
def get_schema(dataset_id: str):
schema = catalog.get_schema(dataset_id)
if not schema:
raise HTTPException(status_code=404, detail="Dataset not found")
return {"id": schema.id, "fields": schema.fields, "description": schema.description}
# 端点:实际查询数据,支持过滤
@app.get("/data/{dataset_id}")
def query_data(dataset_id: str, region: Optional[str] = Query(None)):
df = catalog.load_data(dataset_id) # 实际加载数据
if region:
df = df[df['region'] == region]
return df.to_dict(orient='records')
动态目录(基于文件系统或数据库)
如果数据源是动态增加(如每日新CSV文件),目录不能是写死的,你需要扫描并索引数据源。
import os
import glob
from pathlib import Path
class FileSystemCatalog:
"""基于文件系统的自动发现目录"""
def __init__(self, base_dir: str = "data/"):
self.base_dir = Path(base_dir)
self._index: Dict[str, DatasetSchema] = {}
def refresh_index(self):
"""扫描目录,自动建立索引"""
self._index.clear()
patterns = ["*.csv", "*.parquet", "*.json"]
for pattern in patterns:
for file_path in self.base_dir.rglob(pattern):
dataset_id = str(file_path.relative_to(self.base_dir).with_suffix('')).replace('/', '.')
# 自动推断字段(读取文件头)
fields = self._infer_fields(file_path)
self._index[dataset_id] = DatasetSchema(
id=dataset_id,
name=file_path.stem,
description=f"Auto-discovered from {file_path}",
source_type=file_path.suffix[1:], # "csv", "parquet"
source_path=str(file_path),
fields=fields
)
def _infer_fields(self, file_path: Path) -> List[Dict[str, Any]]:
# 简化推断:读取第一行作为字段名
# 生产环境需使用 pandas 或 pyarrow 推断类型
try:
if file_path.suffix == '.csv':
import csv
with open(file_path) as f:
reader = csv.reader(f)
headers = next(reader)
return [{"name": h, "type": "string"} for h in headers] # 简化
except Exception:
return []
return []
def search(self, keyword: str) -> List[DatasetSchema]:
"""按名称或描述搜索"""
return [ds for ds in self._index.values()
if keyword.lower() in ds.name.lower() or keyword.lower() in ds.description.lower()]
企业级目录(元数据存储 + 血缘)
对于复杂场景,数据目录应该持久化到数据库,并支持:
- 版本管理:数据集的Schema变更。
- 血缘追踪:数据从哪里来(ETL过程)。
- 权限控制:谁可以访问哪些数据集。
- 数据质量指标:空值率、唯一性等。
# 使用 SQLAlchemy 定义模型
from sqlalchemy import Column, String, Integer, JSON, DateTime, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
class DatasetMetadata(Base):
__tablename__ = 'dataset_catalog'
id = Column(Integer, primary_key=True)
dataset_id = Column(String, unique=True, index=True) # e.g. "sales.orders"
name = Column(String)
description = Column(String)
schema_fields = Column(JSON) # 字段列表
source = Column(String) # 数据源信息
owner = Column(String)
created_at = Column(DateTime, default=datetime.utcnow)
last_updated = Column(DateTime)
# 然后你可以通过 ORM 查询,形成 API 目录端点
# GET /catalog/search?q=revenue
# GET /catalog/sales.orders/lineage
最佳实践建议
-
一致性是关键:
- 将目录 API 作为所有数据请求的唯一入口。
- API端点设计:
GET /catalog(列表),GET /catalog/{id}(详情),GET /data/{id}(实际数据)。
-
懒加载与缓存:
- 目录元数据应该常驻内存(或使用Redis缓存)。
- 数据本身(DataFrame)根据场景选择性缓存(适合不常变的数据)。
-
类型与验证:
- 在目录中定义明确的字段类型(
int,float,date,string)。 - 在返回数据前,根据目录定义执行类型转换和校验。
- 在目录中定义明确的字段类型(
-
可发现性:
- 支持(
tags)和全文搜索(search)功能,方便用户找到数据。 - 提供数据预览端点(
GET /catalog/{id}/preview?limit=5)。
- 支持(
-
安全性:
- 不要在目录中直接暴露文件路径或数据库连接串。
- 通过目录ID进行映射,这样底层存储迁移时,API消费者无需变化。
| 维度 | 小型/内存方案 | 大型/持久化方案 |
|---|---|---|
| 存储 | Python字典 / dataclass |
SQL数据库 / NoSQL / 图数据库 |
| 注册方式 | 手动 register() |
自动扫描 / 事件监听 / 手动录入 |
| 灵活性 | 启动时固定 | 运行时动态添加 |
| 适合场景 | 微服务、原型、小团队 | 数据中台、多团队、复杂血缘 |
选择建议:从方案一(内存目录 + FastAPI)开始,满足大部分API数据目录需求,当数据集超过100个、需要多人协作时,再迁移到方案三(数据库持久化)。