RailWise KB已发布

API接口集成模板

RailWise API接口集成标准化代码模板，支持RESTful API设计、数据交换、身份认证、错误处理、限流控制与版本管理，实现与第三方系统的标准化数据对接。

复核 2026-07-09入门公开可引用RailWise 技术团队

template-doc

API接口集成模板

1. 模板概述

1.1 用途

本模板提供RailWise系统与外部系统（如WorkWise、业主平台、政府监管平台等）进行API集成的标准化代码框架，支持：

客户端SDK：作为API消费者，调用外部系统接口
服务端框架：作为API提供者，对外暴露监测数据接口
数据交换：标准化的数据序列化与反序列化
安全认证：OAuth2.0、JWT、API Key等多种认证方式
错误处理：统一的错误码与异常处理机制
限流控制：请求频率限制与熔断保护

1.2 适用场景

场景	角色	协议	数据方向
上报监测数据到监管平台	客户端	HTTPS/POST	出站
接收第三方系统查询请求	服务端	HTTPS/GET	入站
与WorkWise平台双向同步	双向	HTTPS/REST	双向
移动端数据同步	服务端	HTTPS/REST	入站
物联网设备数据接入	服务端	MQTT/HTTP	入站

1.3 输入输出

输入：API请求参数、认证信息、请求体数据
输出：API响应数据、状态码、错误信息
中间产物：请求日志、响应日志、认证令牌

2. 代码示例

2.1 客户端SDK（API消费者）

"""
RailWise API客户端SDK
用于调用外部系统API的标准化客户端

作者: RailWise技术部
版本: 1.0.0
"""

import json
import logging
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union
from urllib.parse import urljoin, urlparse

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("RailWise.APIClient")


class AuthType(Enum):
    """认证类型枚举"""
    NONE = "none"
    API_KEY = "api_key"
    BASIC = "basic"
    BEARER = "bearer"
    OAUTH2 = "oauth2"
    JWT = "jwt"


class HttpMethod(Enum):
    """HTTP方法枚举"""
    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"


@dataclass
class APIConfig:
    """API配置"""
    base_url: str
    auth_type: AuthType = AuthType.NONE
    api_key: Optional[str] = None
    username: Optional[str] = None
    password: Optional[str] = None
    client_id: Optional[str] = None
    client_secret: Optional[str] = None
    token_url: Optional[str] = None
    timeout: int = 30
    max_retries: int = 3
    retry_delay: float = 1.0
    rate_limit_per_second: int = 10
    verify_ssl: bool = True
    custom_headers: Dict[str, str] = field(default_factory=dict)


@dataclass
class APIResponse:
    """API响应"""
    success: bool
    status_code: int = 0
    data: Any = None
    headers: Dict[str, str] = field(default_factory=dict)
    error_message: Optional[str] = None
    request_time_ms: float = 0.0
    retry_count: int = 0


class AuthProvider(ABC):
    """认证提供者抽象基类"""

    @abstractmethod
    def get_auth_header(self) -> Dict[str, str]:
        """获取认证请求头"""
        pass

    @abstractmethod
    def refresh(self) -> bool:
        """刷新认证令牌"""
        pass

    @abstractmethod
    def is_expired(self) -> bool:
        """检查令牌是否过期"""
        pass


class APIKeyAuth(AuthProvider):
    """API Key认证"""

    def __init__(self, api_key: str, header_name: str = "X-API-Key"):
        self.api_key = api_key
        self.header_name = header_name

    def get_auth_header(self) -> Dict[str, str]:
        return {self.header_name: self.api_key}

    def refresh(self) -> bool:
        return True  # API Key不需要刷新

    def is_expired(self) -> bool:
        return False


class BearerAuth(AuthProvider):
    """Bearer Token认证"""

    def __init__(self, token: str):
        self.token = token

    def get_auth_header(self) -> Dict[str, str]:
        return {"Authorization": f"Bearer {self.token}"}

    def refresh(self) -> bool:
        return True

    def is_expired(self) -> bool:
        return False


class OAuth2Auth(AuthProvider):
    """OAuth2.0认证"""

    def __init__(
        self,
        client_id: str,
        client_secret: str,
        token_url: str,
        scope: Optional[str] = None
    ):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = token_url
        self.scope = scope
        self.access_token: Optional[str] = None
        self.refresh_token: Optional[str] = None
        self.expires_at: Optional[datetime] = None
        self._token_lock = False

    def get_auth_header(self) -> Dict[str, str]:
        if self.is_expired():
            self.refresh()

        return {"Authorization": f"Bearer {self.access_token}"} if self.access_token else {}

    def refresh(self) -> bool:
        """刷新OAuth2令牌"""
        try:
            response = requests.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.client_id,
                    "client_secret": self.client_secret,
                    "scope": self.scope
                },
                timeout=30
            )

            if response.status_code == 200:
                token_data = response.json()
                self.access_token = token_data.get("access_token")
                self.refresh_token = token_data.get("refresh_token")

                expires_in = token_data.get("expires_in", 3600)
                self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in - 60)

                logger.info("OAuth2令牌刷新成功")
                return True
            else:
                logger.error(f"OAuth2令牌刷新失败: {response.status_code}")
                return False

        except Exception as e:
            logger.error(f"OAuth2令牌刷新异常: {e}")
            return False

    def is_expired(self) -> bool:
        if not self.expires_at:
            return True
        return datetime.now(timezone.utc) >= self.expires_at


class JWTAuth(AuthProvider):
    """JWT认证"""

    def __init__(self, secret: str, algorithm: str = "HS256", payload: Optional[Dict] = None):
        self.secret = secret
        self.algorithm = algorithm
        self.payload = payload or {}
        self.token: Optional[str] = None
        self.expires_at: Optional[datetime] = None

    def get_auth_header(self) -> Dict[str, str]:
        if self.is_expired():
            self._generate_token()

        return {"Authorization": f"Bearer {self.token}"} if self.token else {}

    def _generate_token(self):
        """生成JWT令牌"""
        try:
            import jwt as pyjwt

            payload = self.payload.copy()
            payload["iat"] = datetime.now(timezone.utc)
            payload["exp"] = datetime.now(timezone.utc) + timedelta(hours=1)

            self.token = pyjwt.encode(payload, self.secret, algorithm=self.algorithm)
            self.expires_at = payload["exp"]

        except ImportError:
            logger.error("PyJWT未安装，请安装: pip install PyJWT")
        except Exception as e:
            logger.error(f"JWT生成失败: {e}")

    def refresh(self) -> bool:
        self._generate_token()
        return self.token is not None

    def is_expired(self) -> bool:
        if not self.expires_at:
            return True
        return datetime.now(timezone.utc) >= self.expires_at


class RateLimiter:
    """请求频率限制器"""

    def __init__(self, max_requests: int, window_seconds: int = 1):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests: List[float] = []

    def acquire(self) -> bool:
        """获取请求许可"""
        now = time.time()

        # 清理过期的请求记录
        self.requests = [t for t in self.requests if now - t < self.window]

        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True

        return False

    def wait_time(self) -> float:
        """计算需要等待的时间"""
        if len(self.requests) < self.max_requests:
            return 0

        now = time.time()
        oldest = min(self.requests)
        return max(0, self.window - (now - oldest))


class CircuitBreaker:
    """熔断器"""

    class State(Enum):
        CLOSED = "closed"       # 正常
        OPEN = "open"           # 熔断
        HALF_OPEN = "half_open"  # 半开

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls

        self.state = self.State.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None

    def can_execute(self) -> bool:
        """检查是否可以执行请求"""
        if self.state == self.State.CLOSED:
            return True

        if self.state == self.State.OPEN:
            if self.last_failure_time and time.time() - self.last_failure_time >= self.recovery_timeout:
                self.state = self.State.HALF_OPEN
                self.success_count = 0
                return True
            return False

        if self.state == self.State.HALF_OPEN:
            return self.success_count < self.half_open_max_calls

        return True

    def record_success(self):
        """记录成功"""
        if self.state == self.State.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.half_open_max_calls:
                self.state = self.State.CLOSED
                self.failure_count = 0
        else:
            self.failure_count = 0

    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()

        if self.state == self.State.HALF_OPEN:
            self.state = self.State.OPEN
        elif self.failure_count >= self.failure_threshold:
            self.state = self.State.OPEN


class RailWiseAPIClient:
    """RailWise API客户端"""

    def __init__(self, config: APIConfig):
        self.config = config
        self.logger = logging.getLogger("RailWiseAPIClient")

        # 创建会话
        self.session = requests.Session()

        # 配置重试策略
        retry_strategy = Retry(
            total=config.max_retries,
            backoff_factor=config.retry_delay,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["HEAD", "GET", "OPTIONS", "POST", "PUT"]
        )

        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

        # 初始化认证
        self.auth_provider = self._create_auth_provider()

        # 初始化限流器
        self.rate_limiter = RateLimiter(
            max_requests=config.rate_limit_per_second,
            window_seconds=1
        )

        # 初始化熔断器
        self.circuit_breaker = CircuitBreaker()

    def _create_auth_provider(self) -> Optional[AuthProvider]:
        """创建认证提供者"""
        if self.config.auth_type == AuthType.API_KEY:
            return APIKeyAuth(self.config.api_key)
        elif self.config.auth_type == AuthType.BEARER:
            return BearerAuth(self.config.api_key)
        elif self.config.auth_type == AuthType.OAUTH2:
            return OAuth2Auth(
                self.config.client_id,
                self.config.client_secret,
                self.config.token_url
            )
        elif self.config.auth_type == AuthType.JWT:
            return JWTAuth(self.config.client_secret)
        elif self.config.auth_type == AuthType.BASIC:
            # Basic Auth通过requests的auth参数处理
            return None
        return None

    def request(
        self,
        method: HttpMethod,
        endpoint: str,
        params: Optional[Dict] = None,
        data: Optional[Dict] = None,
        headers: Optional[Dict] = None,
        files: Optional[Dict] = None,
        timeout: Optional[int] = None
    ) -> APIResponse:
        """
        发送HTTP请求

        Args:
            method: HTTP方法
            endpoint: API端点（相对路径）
            params: URL参数
            data: 请求体数据
            headers: 自定义请求头
            files: 上传文件
            timeout: 超时时间

        Returns:
            API响应
        """
        start_time = time.time()
        response = APIResponse(success=False)

        # 检查熔断器
        if not self.circuit_breaker.can_execute():
            response.error_message = "熔断器已打开，请求被拒绝"
            self.logger.warning(response.error_message)
            return response

        # 限流控制
        while not self.rate_limiter.acquire():
            wait_time = self.rate_limiter.wait_time()
            if wait_time > 0:
                self.logger.debug(f"限流等待: {wait_time:.2f}s")
                time.sleep(wait_time)

        # 构建请求
        url = urljoin(self.config.base_url, endpoint)
        request_headers = self._build_headers(headers)

        try:
            # 发送请求
            request_kwargs = {
                "headers": request_headers,
                "params": params,
                "timeout": timeout or self.config.timeout,
                "verify": self.config.verify_ssl
            }

            # Basic Auth
            if self.config.auth_type == AuthType.BASIC:
                request_kwargs["auth"] = (self.config.username, self.config.password)

            # 请求体
            if data and not files:
                request_kwargs["json"] = data
            elif files:
                request_kwargs["data"] = data
                request_kwargs["files"] = files

            resp = self.session.request(method.value, url, **request_kwargs)

            # 记录响应
            response.status_code = resp.status_code
            response.headers = dict(resp.headers)
            response.request_time_ms = (time.time() - start_time) * 1000

            # 处理响应
            if resp.status_code in range(200, 300):
                response.success = True

                # 解析JSON响应
                try:
                    response.data = resp.json()
                except ValueError:
                    response.data = resp.text

                self.circuit_breaker.record_success()
                self.logger.info(f"请求成功: {method.value} {url} ({response.status_code})")

            else:
                response.error_message = f"HTTP {resp.status_code}: {resp.text[:200]}"
                self.circuit_breaker.record_failure()
                self.logger.error(f"请求失败: {method.value} {url} ({resp.status_code})")

        except requests.exceptions.Timeout:
            response.error_message = "请求超时"
            self.circuit_breaker.record_failure()
            self.logger.error(f"请求超时: {url}")

        except requests.exceptions.ConnectionError:
            response.error_message = "连接错误"
            self.circuit_breaker.record_failure()
            self.logger.error(f"连接错误: {url}")

        except Exception as e:
            response.error_message = f"请求异常: {str(e)}"
            self.circuit_breaker.record_failure()
            self.logger.error(f"请求异常: {url}, {e}")

        return response

    def _build_headers(self, custom_headers: Optional[Dict] = None) -> Dict[str, str]:
        """构建请求头"""
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "RailWise-API-Client/1.0.0"
        }

        # 添加认证头
        if self.auth_provider:
            auth_headers = self.auth_provider.get_auth_header()
            headers.update(auth_headers)

        # 添加自定义头
        if self.config.custom_headers:
            headers.update(self.config.custom_headers)

        if custom_headers:
            headers.update(custom_headers)

        return headers

    # 便捷方法
    def get(self, endpoint: str, params: Optional[Dict] = None, **kwargs) -> APIResponse:
        """GET请求"""
        return self.request(HttpMethod.GET, endpoint, params=params, **kwargs)

    def post(self, endpoint: str, data: Optional[Dict] = None, **kwargs) -> APIResponse:
        """POST请求"""
        return self.request(HttpMethod.POST, endpoint, data=data, **kwargs)

    def put(self, endpoint: str, data: Optional[Dict] = None, **kwargs) -> APIResponse:
        """PUT请求"""
        return self.request(HttpMethod.PUT, endpoint, data=data, **kwargs)

    def delete(self, endpoint: str, **kwargs) -> APIResponse:
        """DELETE请求"""
        return self.request(HttpMethod.DELETE, endpoint, **kwargs)

    def close(self):
        """关闭会话"""
        self.session.close()

2.2 服务端框架（API提供者）

"""
RailWise API服务端框架
基于FastAPI的监测数据API服务

作者: RailWise技术部
版本: 1.0.0
"""

from datetime import datetime, timezone
from enum import Enum
from typing import Any, Dict, List, Optional

from fastapi import FastAPI, HTTPException, Depends, Query, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, validator

# 创建FastAPI应用
app = FastAPI(
    title="RailWise Monitoring API",
    description="RailWise监测数据开放平台API",
    version="1.0.0",
    docs_url="/api/docs",
    redoc_url="/api/redoc"
)

# 中间件
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 安全认证
security = HTTPBearer()


# 数据模型
class MonitoringRecord(BaseModel):
    """监测记录模型"""
    record_id: str = Field(..., description="记录唯一标识")
    project_id: str = Field(..., description="项目编号")
    point_id: str = Field(..., description="测点编号")
    point_name: Optional[str] = Field(None, description="测点名称")
    observation_time: datetime = Field(..., description="观测时间")
    value_x: Optional[float] = Field(None, description="X方向位移(mm)")
    value_y: Optional[float] = Field(None, description="Y方向位移(mm)")
    value_z: Optional[float] = Field(None, description="Z方向位移(mm)")
    value_d: Optional[float] = Field(None, description="累计位移(mm)")
    alert_level: Optional[str] = Field("正常", description="预警等级")

    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }


class MonitoringDataQuery(BaseModel):
    """监测数据查询参数"""
    project_id: str = Field(..., description="项目编号")
    point_id: Optional[str] = Field(None, description="测点编号")
    start_time: Optional[datetime] = Field(None, description="开始时间")
    end_time: Optional[datetime] = Field(None, description="结束时间")
    alert_level: Optional[str] = Field(None, description="预警等级筛选")
    page: int = Field(1, ge=1, description="页码")
    page_size: int = Field(100, ge=1, le=1000, description="每页数量")

    @validator('end_time')
    def check_time_range(cls, v, values):
        if v and values.get('start_time') and v < values['start_time']:
            raise ValueError('结束时间必须大于开始时间')
        return v


class APIResponse(BaseModel):
    """API统一响应"""
    code: int = Field(200, description="状态码")
    message: str = Field("success", description="消息")
    data: Any = Field(None, description="数据")
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    request_id: Optional[str] = Field(None, description="请求ID")


class ErrorResponse(BaseModel):
    """错误响应"""
    code: int = Field(..., description="错误码")
    message: str = Field(..., description="错误消息")
    detail: Optional[str] = Field(None, description="详细错误信息")
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


# 认证验证
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str:
    """验证JWT令牌"""
    token = credentials.credentials

    # 实际项目中应验证JWT签名和过期时间
    # 这里简化处理
    if not token or token == "invalid":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证令牌",
            headers={"WWW-Authenticate": "Bearer"}
        )

    return token


# 模拟数据存储（实际项目中使用数据库）
MOCK_DATA: List[Dict] = [
    {
        "record_id": "REC001",
        "project_id": "RW-2024-001",
        "point_id": "JC-01",
        "point_name": "3号线K12+350左线",
        "observation_time": datetime.now(timezone.utc),
        "value_x": 0.15,
        "value_y": -0.05,
        "value_z": 0.01,
        "value_d": 2.35,
        "alert_level": "正常"
    }
]


# API端点
@app.get("/api/v1/health", response_model=APIResponse, tags=["系统"])
async def health_check():
    """健康检查"""
    return APIResponse(
        code=200,
        message="服务运行正常",
        data={"status": "healthy", "version": "1.0.0"}
    )


@app.get("/api/v1/monitoring/data", response_model=APIResponse, tags=["监测数据"])
async def get_monitoring_data(
    project_id: str = Query(..., description="项目编号"),
    point_id: Optional[str] = Query(None, description="测点编号"),
    start_time: Optional[datetime] = Query(None, description="开始时间"),
    end_time: Optional[datetime] = Query(None, description="结束时间"),
    alert_level: Optional[str] = Query(None, description="预警等级"),
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(100, ge=1, le=1000, description="每页数量"),
    token: str = Depends(verify_token)
):
    """
    查询监测数据

    支持按项目、测点、时间范围、预警等级等条件查询监测数据
    """
    try:
        # 筛选数据
        data = MOCK_DATA.copy()

        if project_id:
            data = [d for d in data if d["project_id"] == project_id]

        if point_id:
            data = [d for d in data if d["point_id"] == point_id]

        if start_time:
            data = [d for d in data if d["observation_time"] >= start_time]

        if end_time:
            data = [d for d in data if d["observation_time"] <= end_time]

        if alert_level:
            data = [d for d in data if d["alert_level"] == alert_level]

        # 分页
        total = len(data)
        start_idx = (page - 1) * page_size
        end_idx = start_idx + page_size
        paginated_data = data[start_idx:end_idx]

        return APIResponse(
            code=200,
            message="success",
            data={
                "total": total,
                "page": page,
                "page_size": page_size,
                "records": paginated_data
            }
        )

    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询失败: {str(e)}"
        )


@app.post("/api/v1/monitoring/data", response_model=APIResponse, tags=["监测数据"])
async def create_monitoring_data(
    record: MonitoringRecord,
    token: str = Depends(verify_token)
):
    """
    创建监测数据记录

    接收新的监测数据记录并存储
    """
    try:
        # 验证数据
        if not record.project_id or not record.point_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="项目编号和测点编号不能为空"
            )

        # 存储数据（实际项目中写入数据库）
        record_dict = record.dict()
        MOCK_DATA.append(record_dict)

        return APIResponse(
            code=201,
            message="数据创建成功",
            data={"record_id": record.record_id}
        )

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"创建失败: {str(e)}"
        )


@app.get("/api/v1/monitoring/points", response_model=APIResponse, tags=["监测点"])
async def get_monitoring_points(
    project_id: str = Query(..., description="项目编号"),
    token: str = Depends(verify_token)
):
    """获取监测点列表"""
    try:
        # 从数据中提取测点信息
        points = {}
        for record in MOCK_DATA:
            if record["project_id"] == project_id:
                point_id = record["point_id"]
                if point_id not in points:
                    points[point_id] = {
                        "point_id": point_id,
                        "point_name": record.get("point_name", ""),
                        "latest_value": record.get("value_d"),
                        "latest_time": record.get("observation_time"),
                        "alert_level": record.get("alert_level", "正常")
                    }

        return APIResponse(
            code=200,
            message="success",
            data=list(points.values())
        )

    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询失败: {str(e)}"
        )


@app.get("/api/v1/monitoring/alerts", response_model=APIResponse, tags=["预警"])
async def get_active_alerts(
    project_id: str = Query(..., description="项目编号"),
    alert_level: Optional[str] = Query(None, description="预警等级"),
    token: str = Depends(verify_token)
):
    """获取活跃预警列表"""
    try:
        alerts = [
            record for record in MOCK_DATA
            if record["project_id"] == project_id
            and record.get("alert_level", "正常") != "正常"
        ]

        if alert_level:
            alerts = [a for a in alerts if a["alert_level"] == alert_level]

        return APIResponse(
            code=200,
            message="success",
            data=alerts
        )

    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询失败: {str(e)}"
        )


# 错误处理
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    """HTTP异常处理"""
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            code=exc.status_code,
            message=exc.detail,
            detail=str(exc)
        ).dict()
    )


@app.exception_handler(Exception)
async def general_exception_handler(request, exc):
    """通用异常处理"""
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content=ErrorResponse(
            code=500,
            message="服务器内部错误",
            detail=str(exc)
        ).dict()
    )


# 启动服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. 参数说明

3.1 APIConfig 客户端配置参数

参数名	类型	必填	默认值	说明
`base_url`	str	是	-	API基础URL
`auth_type`	AuthType	否	NONE	认证类型
`api_key`	str	否	None	API密钥
`username`	str	否	None	用户名（Basic Auth）
`password`	str	否	None	密码
`client_id`	str	否	None	OAuth2客户端ID
`client_secret`	str	否	None	OAuth2客户端密钥
`token_url`	str	否	None	OAuth2令牌URL
`timeout`	int	否	30	请求超时(秒)
`max_retries`	int	否	3	最大重试次数
`retry_delay`	float	否	1.0	重试延迟(秒)
`rate_limit_per_second`	int	否	10	每秒请求限制
`verify_ssl`	bool	否	True	验证SSL证书
`custom_headers`	dict	否	{}	自定义请求头

3.2 MonitoringRecord 数据模型参数

参数名	类型	必填	说明
`record_id`	str	是	记录唯一标识
`project_id`	str	是	项目编号
`point_id`	str	是	测点编号
`point_name`	str	否	测点名称
`observation_time`	datetime	是	观测时间
`value_x`	float	否	X方向位移(mm)
`value_y`	float	否	Y方向位移(mm)
`value_z`	float	否	Z方向位移(mm)
`value_d`	float	否	累计位移(mm)
`alert_level`	str	否	预警等级

4. 使用示例

4.1 客户端调用示例

from template_api_integration import (
    RailWiseAPIClient, APIConfig, AuthType, HttpMethod
)

# 1. 配置API客户端
config = APIConfig(
    base_url="https://api.workwise.railwise.com",
    auth_type=AuthType.API_KEY,
    api_key="your-api-key-here",
    timeout=30,
    max_retries=3,
    rate_limit_per_second=10
)

# 2. 创建客户端
client = RailWiseAPIClient(config)

# 3. GET请求 - 查询监测数据
response = client.get(
    "/api/v1/monitoring/data",
    params={
        "project_id": "RW-2024-001",
        "point_id": "JC-01",
        "page": 1,
        "page_size": 100
    }
)

if response.success:
    print(f"查询成功: {response.data}")
else:
    print(f"查询失败: {response.error_message}")

# 4. POST请求 - 上报监测数据
new_record = {
    "record_id": "REC002",
    "project_id": "RW-2024-001",
    "point_id": "JC-02",
    "point_name": "3号线K12+350右线",
    "observation_time": "2024-07-08T12:00:00Z",
    "value_x": 0.22,
    "value_y": -0.03,
    "value_z": 0.01,
    "value_d": 3.12,
    "alert_level": "正常"
}

response = client.post(
    "/api/v1/monitoring/data",
    data=new_record
)

if response.success:
    print(f"上报成功: {response.data}")
else:
    print(f"上报失败: {response.error_message}")

# 5. 关闭客户端
client.close()

4.2 OAuth2认证示例

from template_api_integration import RailWiseAPIClient, APIConfig, AuthType

# OAuth2配置
config = APIConfig(
    base_url="https://api.regulatory.gov.cn",
    auth_type=AuthType.OAUTH2,
    client_id="your-client-id",
    client_secret="your-client-secret",
    token_url="https://auth.regulatory.gov.cn/oauth2/token"
)

client = RailWiseAPIClient(config)

# 令牌会自动获取和刷新
response = client.get("/api/v1/projects/RW-2024-001/status")

if response.success:
    print(f"查询成功: {response.data}")

client.close()

4.3 批量上报数据

from template_api_integration import RailWiseAPIClient, APIConfig, AuthType
import time

# 批量上报监测数据
config = APIConfig(
    base_url="https://api.workwise.railwise.com",
    auth_type=AuthType.API_KEY,
    api_key="your-api-key",
    rate_limit_per_second=5  # 限制每秒5请求
)

client = RailWiseAPIClient(config)

records = [
    {"record_id": f"REC_{i:03d}", "project_id": "RW-2024-001", "point_id": "JC-01", "value_d": i * 0.1}
    for i in range(100)
]

success_count = 0
for record in records:
    response = client.post("/api/v1/monitoring/data/batch", data=record)

    if response.success:
        success_count += 1
    else:
        print(f"上报失败: {record['record_id']}, {response.error_message}")

    # 限流自动处理

print(f"批量上报完成: {success_count}/{len(records)} 成功")
client.close()

4.4 启动API服务端

# 保存为 api_server.py
from template_api_integration import app
import uvicorn

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        ssl_keyfile="server.key",
        ssl_certfile="server.crt"
    )

启动服务：

python api_server.py

访问API文档：

https://localhost:8000/api/docs

5. 输出示例

5.1 API响应示例

{
  "code": 200,
  "message": "success",
  "data": {
    "total": 156,
    "page": 1,
    "page_size": 100,
    "records": [
      {
        "record_id": "REC001",
        "project_id": "RW-2024-001",
        "point_id": "JC-01",
        "point_name": "3号线K12+350左线",
        "observation_time": "2024-07-08T08:00:00+00:00",
        "value_x": 0.15,
        "value_y": -0.05,
        "value_z": 0.01,
        "value_d": 2.35,
        "alert_level": "正常"
      }
    ]
  },
  "timestamp": "2024-07-08T12:00:00.123456+00:00",
  "request_id": "req-1234567890"
}

5.2 错误响应示例

{
  "code": 401,
  "message": "无效的认证令牌",
  "detail": "Bearer token验证失败",
  "timestamp": "2024-07-08T12:00:00.123456+00:00"
}

5.3 请求日志示例

2024-07-08 12:00:05,123 - RailWiseAPIClient - INFO - 请求成功: GET https://api.workwise.railwise.com/api/v1/monitoring/data (200), 耗时: 125.67ms
2024-07-08 12:00:06,456 - RailWiseAPIClient - ERROR - 请求失败: POST https://api.workwise.railwise.com/api/v1/monitoring/data (500), 耗时: 234.56ms

6. 扩展指南

6.1 添加自定义认证方式

from template_api_integration import AuthProvider

class HMACAuth(AuthProvider):
    """HMAC签名认证"""

    def __init__(self, api_key: str, api_secret: str):
        self.api_key = api_key
        self.api_secret = api_secret

    def get_auth_header(self) -> Dict[str, str]:
        import hmac
        import hashlib
        import base64

        timestamp = str(int(time.time()))
        message = f"{timestamp}{self.api_key}"
        signature = base64.b64encode(
            hmac.new(
                self.api_secret.encode(),
                message.encode(),
                hashlib.sha256
            ).digest()
        ).decode()

        return {
            "X-API-Key": self.api_key,
            "X-Timestamp": timestamp,
            "X-Signature": signature
        }

    def refresh(self) -> bool:
        return True

    def is_expired(self) -> bool:
        return False

6.2 添加自定义中间件

from fastapi import Request
import time

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    """请求日志中间件"""
    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time

    logger.info(
        f"{request.method} {request.url.path} "
        f"- {response.status_code} "
        f"- {process_time:.3f}s"
    )

    response.headers["X-Process-Time"] = str(process_time)
    return response

6.3 集成到监测工作流

from template_data_import import MonitoringRecord
from template_api_integration import RailWiseAPIClient, APIConfig, AuthType

class MonitoringDataUploader:
    """监测数据上传器"""

    def __init__(self, api_client: RailWiseAPIClient):
        self.client = api_client

    def upload_record(self, record: MonitoringRecord) -> bool:
        """上传单条记录"""
        response = self.client.post(
            "/api/v1/monitoring/data",
            data=record.dict()
        )
        return response.success

    def upload_batch(self, records: List[MonitoringRecord]) -> Tuple[int, int]:
        """批量上传记录"""
        success = 0
        failed = 0

        for record in records:
            if self.upload_record(record):
                success += 1
            else:
                failed += 1

        return success, failed

6.4 缓存API响应

import functools
import time

class CachedAPIClient(RailWiseAPIClient):
    """带缓存的API客户端"""

    def __init__(self, config: APIConfig, cache_ttl: int = 300):
        super().__init__(config)
        self.cache = {}
        self.cache_ttl = cache_ttl

    def get(self, endpoint: str, params: Optional[Dict] = None, **kwargs) -> APIResponse:
        """带缓存的GET请求"""
        cache_key = f"{endpoint}:{json.dumps(params, sort_keys=True)}"

        # 检查缓存
        if cache_key in self.cache:
            cached_response, cached_time = self.cache[cache_key]
            if time.time() - cached_time < self.cache_ttl:
                self.logger.debug(f"缓存命中: {cache_key}")
                return cached_response

        # 发送请求
        response = super().get(endpoint, params, **kwargs)

        # 缓存成功响应
        if response.success:
            self.cache[cache_key] = (response, time.time())

        return response

7. 常见问题

Q1: 如何处理API限流？

A: 客户端内置了限流器和重试机制：

config = APIConfig(
    rate_limit_per_second=10,  # 每秒最多10请求
    max_retries=3,             # 失败重试3次
    retry_delay=2.0            # 重试间隔2秒
)

Q2: 如何调试API请求？

A: 启用请求日志和响应日志：

import logging

# 启用requests库调试日志
logging.getLogger("urllib3").setLevel(logging.DEBUG)

# 客户端日志
client.logger.setLevel(logging.DEBUG)

Q3: 如何支持异步请求？

A: 使用aiohttp实现异步客户端：

import aiohttp
import asyncio

class AsyncAPIClient:
    async def get(self, endpoint, params=None):
        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as response:
                return await response.json()

Q4: 如何部署API服务？

A: 使用Docker部署：

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "api_server:app", "--host", "0.0.0.0", "--port", "8000"]

Q5: 如何生成API文档？

A: FastAPI自动生成交互式API文档：

Swagger UI: /api/docs
ReDoc: /api/redoc

元数据标签

ai_tags:
  domain: ["工程监测", "系统集成", "API开发"]
  technology: ["Python", "FastAPI", "HTTP", "RESTful", "OAuth2", "JWT"]
  standard: ["GB 50497", "CJJ/T 202"]
  product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"]
  difficulty: "advanced"
  role: ["后端开发工程师", "系统集成工程师", "API设计师"]
  task_type: ["API开发", "系统集成", "数据交换"]

API接口集成模板

API接口集成模板

1. 模板概述

1.1 用途

1.2 适用场景

1.3 输入输出

2. 代码示例

2.1 客户端SDK（API消费者）

2.2 服务端框架（API提供者）

3. 参数说明

3.1 APIConfig 客户端配置参数

3.2 MonitoringRecord 数据模型参数

4. 使用示例

4.1 客户端调用示例

4.2 OAuth2认证示例

4.3 批量上报数据

4.4 启动API服务端

5. 输出示例

5.1 API响应示例

5.2 错误响应示例

5.3 请求日志示例

6. 扩展指南

6.1 添加自定义认证方式

6.2 添加自定义中间件

6.3 集成到监测工作流

6.4 缓存API响应

7. 常见问题

Q1: 如何处理API限流？

Q2: 如何调试API请求？

Q3: 如何支持异步请求？

Q4: 如何部署API服务？

Q5: 如何生成API文档？

相关文档链接

元数据标签