基础设施层

services/infra/ 包含系统的底层支撑服务。

---

模块列表

模块	文件	功能	状态
数据库	database.py	asyncpg 连接池、CRUD	✅
缓存	cache.py	Redis 缓存服务	✅
日志	logger.py	日志管理	✅
安全	security.py	API 密钥、密码哈希	✅
认证依赖	auth_dep.py	JWT 认证依赖注入	✅
认证中间件	auth_middleware.py	认证中间件	✅
验证器	validators.py	输入验证	✅
通知	notification.py	消息通知	✅
分享	share.py	分享链接管理	✅

---

1. 数据库 (database.py)

功能

PostgreSQL 数据库连接池管理和 CRUD 操作。

流程

启动 → 连接池初始化 → 请求处理 → 连接复用 → 关闭

核心函数

from services.infra.database import (
    init_db,           # 初始化连接池
    close_db,          # 关闭连接池
    db_health_check,   # 健康检查
    get_psycopg2_connection  # 同步连接（备用）
)

# 初始化（main.py lifespan 中调用）
await init_db()

# 健康检查
health = await db_health_check()
# 返回: {"available": true, "latency_ms": 1.12, "pool": {...}}

连接池配置

_pool = await asyncpg.create_pool(
    host="172.18.0.3",
    port=5432,
    user="bioinfo",
    password="bioinfo",
    database="bioinfo",
    min_size=2,
    max_size=10,
    command_timeout=60,
    timeout=30
)

数据库操作

# 增
patient_id = await create_patient({
    "name": "张三",
    "age": 45,
    "gender": "male"
})

# 查
patient = await get_patient(patient_id)
patients = await list_patients(limit=100, offset=0)

# 改
ok = await update_patient(patient_id, {"age": 46})

# 初始化表
await create_tables()

---

2. 缓存 (cache.py)

功能

Redis 缓存服务，提升 API 响应速度。

核心类

from services.infra.cache import CacheService, get_cache, cached, invalidate_cache

# 获取缓存服务
cache = get_cache()

# 设置缓存
await cache.set("user:123", user_data, ttl=3600)

# 获取缓存
data = await cache.get("user:123")

# 删除缓存
await cache.delete("user:123")

# 模式删除
await invalidate_cache("user:*")

# 装饰器方式
@cached(key_prefix="analysis", ttl=1800)
async def expensive_analysis(data):
    # ...
    return result

缓存策略

数据类型	TTL	说明
用户会话	30 分钟	频繁访问
分析结果	1 小时	计算成本高
统计聚合	5 分钟	实时性要求
静态配置	24 小时	几乎不变

---

3. 日志 (logger.py)

功能

统一日志管理，支持多级别、多输出。

核心函数

from services.infra.logger import get_logger, get_app_logger, get_security_logger

# 获取应用日志
app_logger = get_app_logger()
app_logger.info("服务启动成功")

# 获取安全日志
security_logger = get_security_logger()
security_logger.warning("登录失败: user=admin, ip=192.168.1.1")

# 自定义日志器
logger = get_logger("bioinfo.analysis")
logger.debug("开始分析样本 S001")

日志级别

• DEBUG: 详细调试信息
• INFO: 一般信息
• WARNING: 警告
• ERROR: 错误
• CRITICAL: 严重错误

---

4. 安全 (security.py)

功能

安全工具：密码哈希、API 密钥、限流。

核心函数

from services.infra.security import (
    generate_api_key,
    verify_api_key,
    revoke_api_key,
    hash_password,
    verify_password,
    RateLimiter
)

# API 密钥
key_id, secret = generate_api_key(prefix="bio")
# key_id = "bio_abc123", secret = "sk_xyz..."

is_valid = verify_api_key(key_id, secret)
revoke_api_key(key_id)

# 密码哈希
hashed, salt = hash_password("myPassword123")
is_match = verify_password("myPassword123", hashed, salt)

# 限流
limiter = RateLimiter(max_requests=100, window=60)
allowed, info = limiter.check_rate_limit("user:123")

---

5. 认证 (auth_dep.py)

功能

JWT 认证依赖注入，保护 API 端点。

核心函数

from services.infra.auth_dep import (
    get_current_user,
    require_admin,
    require_doctor_or_admin,
    optional_auth
)

# 获取当前用户（自动验证 JWT）
@router.get("/profile")
async def get_profile(user: dict = Depends(get_current_user)):
    return user

# 需要管理员权限
@router.delete("/admin/users/{user_id}")
async def delete_user(
    user_id: str,
    _: dict = Depends(require_admin)
):
    ...

# 医生或管理员
@router.post("/medical/report")
async def create_report(
    data: ReportData,
    user: dict = Depends(require_doctor_or_admin)
):
    ...

JWT 结构

{
    "sub": "user_id",
    "username": "doctor_wang",
    "role": "doctor",
    "exp": 1716134400  # 过期时间
}

---

6. 验证器 (validators.py)

功能

输入验证和清理。

核心函数

from services.infra.validators import (
    sanitize_filename,
    validate_file_extension,
    validate_content_type,
    validate_sample_id
)

# 清理文件名
safe_name = sanitize_filename("sample_01.fastq.gz")
# -> "sample_01.fastq.gz"

safe_name = sanitize_filename("../../../etc/passwd")
# -> "" (危险字符被移除)

# 验证文件扩展名
is_valid = validate_file_extension("data.fasta", [".fasta", ".fa", ".fq"])

# 验证样本 ID
sample_id = validate_sample_id("S-001-2024")
# -> "S-001-2024" (合法) 或抛出异常

---

7. 通知 (notification.py)

功能

消息通知系统，支持邮件和 WebSocket。

核心函数

from services.infra.notification import (
    NotificationType,
    add_notification,
    send_email_notification,
    get_user_notifications,
    mark_notification_read
)

# 添加通知
notif_id = add_notification({
    "user_id": "user123",
    "type": NotificationType.ANALYSIS_COMPLETE,
    "title": "分析完成",
    "message": "16S 分析已完成，报告已生成",
    "priority": NotificationPriority.HIGH
})

# 获取用户通知
notifications = get_user_notifications(
    user_id="user123",
    unread_only=True,
    limit=50
)

# 标记已读
mark_notification_read(notif_id)

# 发送邮件
send_email_notification(
    to="user@example.com",
    subject="分析报告已生成",
    body="请查看附件报告..."
)

---

8. 分享 (share.py)

功能

分享链接管理，支持密码保护和过期时间。

核心函数

from services.infra.share import (
    create_share_token,
    verify_and_access,
    get_share_info,
    revoke_share_token,
    list_user_shares
)

# 创建分享
token = create_share_token(
    data={"report_id": "R001", "patient_name": "张三"},
    created_by="doctor_wang",
    password="123456",  # 可选密码
    expires_in=7 * 24 * 3600  # 7 天
)

# 访问分享（需要密码）
info = verify_and_access(token, password="123456")

# 查看分享信息
share_info = get_share_info(token)

# 撤销分享
revoke_share_token(token, created_by="doctor_wang")

# 列出用户的分享
shares = list_user_shares(created_by="doctor_wang", status="active")