jiawei/central-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 中心平台

Browse Source
This commit is contained in:
jiawei
2026-09-05 00:40:30 +08:00
commit 33187293b5
37 changed files with 3083 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+23
View File
@@ -0,0 +1,23 @@
# 运行环境标识，development、testing 或 production
ENVIRONMENT=development
# MySQL 连接地址，mysql+asyncmy://用户名:密码@主机:端口/数据库名
DATABASE_URL=mysql+asyncmy://center:change-me@127.0.0.1:3306/center
# Redis 连接地址，末尾数字表示使用的 Redis 数据库编号
REDIS_URL=redis://127.0.0.1:6379/0
# Redis 键前缀，只允许字母、数字、短横线和下划线
REDIS_KEY_PREFIX=center
# Token 有效期，单位为秒且不能小于 60，默认 24 小时
TOKEN_EXPIRE_SECONDS=86400
# 允许 from_url 使用的来源，多个来源使用英文逗号分隔且不要包含路径
TRUSTED_REDIRECT_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
# 允许跨域访问接口的来源，多个来源使用英文逗号分隔
CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000
# 应用日志级别，DEBUG、INFO、WARNING、ERROR 或 CRITICAL
LOG_LEVEL=INFO
.gitignore
+14
View File
@@ -0,0 +1,14 @@
.env
.idea/
.venv/
.uv-cache/
__pycache__/
*.py[cod]
.pytest_cache/
.ruff_cache/
.coverage
htmlcov/
tests/
docs/
AGENTS.md
README.md
+95
View File
@@ -0,0 +1,95 @@
# Center Platform
轻量级统一认证中心，为多个业务系统提供账号管理、登录认证与会话验证 API。
## 特性
- FastAPI 异步接口与统一响应协议
- MySQL 持久化用户数据，Argon2id 安全存储密码
- Redis 管理具有过期时间的 Opaque Token 会话
- 支持用户名或邮箱登录
- 校验受信任的 `from_url`，便于登录后返回来源系统
- Alembic 数据库迁移、pytest 测试与 Ruff 代码检查
## 技术栈
Python 3.11+ · FastAPI · SQLAlchemy 2.0 · MySQL · Redis · Alembic · uv
## 快速开始
确保本机已安装 MySQL 8+、Redis 6+ 和 [uv](https://docs.astral.sh/uv/)，然后执行：
```powershell
uv sync
Copy-Item .env.example .env
uv run alembic upgrade head
uv run fastapi dev app/main.py
```
启动前请在 `.env` 中配置数据库连接、Redis 连接及可信回跳来源。`DATABASE_URL`、`REDIS_URL` 和
`REDIS_KEY_PREFIX` 为必填项，缺失时应用会拒绝启动。完整配置和注释见 [`.env.example`](.env.example)。
服务默认运行在 `http://127.0.0.1:8000`，交互式 API 文档 `http://127.0.0.1:8000/docs`
## API
| 方法 | 路径 | 说明 |
|--------|----------------------|--------------------|
| `GET` | `/health` | 检查服务及依赖状态 |
| `POST` | `/api/auth/register` | 注册用户 |
| `POST` | `/api/auth/login` | 登录并签发 Token |
| `POST` | `/api/auth/verify` | 验证 Token |
| `POST` | `/api/auth/logout` | 注销当前 Token |
| `GET` | `/api/users/me` | 获取当前用户信息 |
登录示例：
```bash
curl -X POST http://127.0.0.1:8000/api/auth/login \
-H "Content-Type: application/json" \
-d '{"account":"alice","password":"password123","from_url":"https://app.example.com/callback"}'
```
受保护接口使用 `Authorization: Bearer <token>` 请求头。
## 响应协议
所有接口返回统一结构：
```json
{
"code": 0,
"data": {},
"msg": "请求成功"
}
```
成功响应使用 HTTP 200 和 `code=0`。失败时 HTTP 状态码与 `code` 一致，`data` 为 `null`。当前错误码包括 `400`、`401`、`403`、
`429` 和 `500`。
## 开发
```powershell
uv run pytest
uv run ruff check app migrations tests
uv run ruff format --check app migrations tests
```
创建数据库迁移：
```powershell
uv run alembic revision --autogenerate -m "描述"
uv run alembic upgrade head
```
## 项目结构
```text
app/api/ HTTP 接口与依赖
app/service/ 业务逻辑
app/repository/ 数据访问
app/model/ SQLAlchemy 模型
app/schema/ 请求与响应模型
app/core/ 配置、数据库、Redis 与安全组件
migrations/ Alembic 数据库迁移
```
alembic.ini
+39
View File
@@ -0,0 +1,39 @@
[alembic]
script_location = migrations
prepend_sys_path = .
path_separator = os
sqlalchemy.url = mysql+asyncmy://center:center@127.0.0.1:3306/center
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
app/__init__.py
View File
app/api/__init__.py
View File
app/api/auth.py
+67
View File
@@ -0,0 +1,67 @@
# 定义认证相关 HTTP 接口
from typing import Annotated
from fastapi import APIRouter, Depends
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.dependencies import get_bearer_token, get_current_user
from app.common.response import ApiResponse, success_response
from app.core.config import Settings, get_settings
from app.core.database import get_db
from app.core.redis import get_redis
from app.model.user import User
from app.schema.auth import LoginRequest, LoginResponse
from app.schema.user import UserRegisterRequest, UserResponse
from app.service.auth_service import AuthService
from app.service.token_service import TokenService
from app.service.user_service import UserService
router = APIRouter(prefix="/api/auth", tags=["认证"])
@router.post("/register", response_model=ApiResponse[UserResponse])
async def register_user(
request: UserRegisterRequest,
session: Annotated[AsyncSession, Depends(get_db)],
) -> ApiResponse[UserResponse]:
"""注册新的中心平台用户"""
user = await UserService(session).register(request)
return success_response(UserResponse.model_validate(user))
@router.post("/login", response_model=ApiResponse[LoginResponse])
async def login_user(
request: LoginRequest,
session: Annotated[AsyncSession, Depends(get_db)],
redis_client: Annotated[Redis, Depends(get_redis)],
settings: Annotated[Settings, Depends(get_settings)],
) -> ApiResponse[LoginResponse]:
"""登录并签发新的 Opaque Token"""
result = await AuthService(session, redis_client, settings).login(request)
return success_response(result)
@router.post("/verify", response_model=ApiResponse[UserResponse])
async def verify_token(
current_user: Annotated[User, Depends(get_current_user)],
) -> ApiResponse[UserResponse]:
"""验证 Token 并返回当前用户身份"""
return success_response(UserResponse.model_validate(current_user))
@router.post("/logout", response_model=ApiResponse[dict[str, object]])
async def logout_user(
token: Annotated[str, Depends(get_bearer_token)],
_current_user: Annotated[User, Depends(get_current_user)],
redis_client: Annotated[Redis, Depends(get_redis)],
settings: Annotated[Settings, Depends(get_settings)],
) -> ApiResponse[dict[str, object]]:
"""注销当前请求使用的 Token"""
await TokenService(
redis_client,
settings.token_expire_seconds,
settings.redis_key_prefix,
).delete(token)
return success_response()
app/api/dependencies.py
+36
View File
@@ -0,0 +1,36 @@
# 提供 API 鉴权与资源依赖
from typing import Annotated
from fastapi import Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import Settings, get_settings
from app.core.database import get_db
from app.core.exceptions import BizException
from app.core.redis import get_redis
from app.model.user import User
from app.service.auth_service import AuthService
bearer_scheme = HTTPBearer(auto_error=False)
async def get_bearer_token(
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
) -> str:
"""从请求头中提取 Bearer Token"""
if credentials is None or credentials.scheme.casefold() != "bearer":
raise BizException(401, "登录已失效")
return credentials.credentials
async def get_current_user(
token: Annotated[str, Depends(get_bearer_token)],
session: Annotated[AsyncSession, Depends(get_db)],
redis_client: Annotated[Redis, Depends(get_redis)],
settings: Annotated[Settings, Depends(get_settings)],
) -> User:
"""验证请求 Token 并提供当前用户"""
return await AuthService(session, redis_client, settings).authenticate(token)
app/api/router.py
+29
View File
@@ -0,0 +1,29 @@
# 汇总应用 API 路由
from typing import Annotated
from fastapi import APIRouter, Depends
from redis.asyncio import Redis
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.auth import router as auth_router
from app.api.user import router as user_router
from app.common.response import ApiResponse, success_response
from app.core.database import get_db
from app.core.redis import get_redis
api_router = APIRouter()
api_router.include_router(auth_router)
api_router.include_router(user_router)
@api_router.get("/health", response_model=ApiResponse[dict[str, str]], tags=["系统"])
async def health_check(
session: Annotated[AsyncSession, Depends(get_db)],
redis_client: Annotated[Redis, Depends(get_redis)],
) -> ApiResponse[dict[str, str]]:
"""检查应用及外部依赖的健康状态"""
await session.execute(text("SELECT 1"))
await redis_client.ping()
return success_response({"status": "ok", "database": "ok", "redis": "ok"})
app/api/user.py
+20
View File
@@ -0,0 +1,20 @@
# 定义用户相关 HTTP 接口
from typing import Annotated
from fastapi import APIRouter, Depends
from app.api.dependencies import get_current_user
from app.common.response import ApiResponse, success_response
from app.model.user import User
from app.schema.user import UserResponse
router = APIRouter(prefix="/api/users", tags=["用户"])
@router.get("/me", response_model=ApiResponse[UserResponse])
async def get_me(
current_user: Annotated[User, Depends(get_current_user)],
) -> ApiResponse[UserResponse]:
"""返回当前登录用户的公开资料"""
return success_response(UserResponse.model_validate(current_user))
app/common/__init__.py
View File
app/common/response.py
+36
View File
@@ -0,0 +1,36 @@
# 定义统一 API 响应结构
from typing import Any, Generic, TypeVar
from pydantic import BaseModel, field_validator
DataT = TypeVar("DataT")
ResponseData = BaseModel | dict[str, Any] | list[Any] | None
class ApiResponse(BaseModel, Generic[DataT]):
code: int = 0
data: DataT | None = None
msg: str = "请求成功"
@field_validator("data")
@classmethod
def validate_data_shape(cls, value: DataT | None) -> DataT | None:
"""限制响应数据为对象、数组或空值"""
if value is not None and not isinstance(value, (BaseModel, dict, list)):
raise ValueError("响应 data 必须为对象、数组或 null")
return value
def success_response(
data: DataT | None = None,
msg: str = "请求成功",
) -> ApiResponse[DataT | dict[str, Any]]:
"""构造统一成功响应"""
response_data: DataT | dict[str, Any] = {} if data is None else data
return ApiResponse(code=0, data=response_data, msg=msg)
def error_content(code: int, msg: str = "请求失败") -> dict[str, ResponseData | int | str]:
"""构造统一失败响应内容"""
return ApiResponse[ResponseData](code=code, data=None, msg=msg).model_dump(mode="json")
app/core/__init__.py
View File
app/core/config.py
+50
View File
@@ -0,0 +1,50 @@
# 管理应用环境配置
from functools import lru_cache
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
def _split_origins(value: str) -> list[str]:
"""拆分并清理逗号分隔的来源配置"""
return [origin.strip().rstrip("/") for origin in value.split(",") if origin.strip()]
class Settings(BaseSettings):
environment: str = "development"
database_url: str = Field(min_length=1)
redis_url: str = Field(min_length=1)
redis_key_prefix: str = Field(
min_length=1,
max_length=64,
pattern=r"^[A-Za-z0-9][A-Za-z0-9_-]*$",
)
token_expire_seconds: int = Field(default=86400, ge=60)
trusted_redirect_origins: str = ""
cors_origins: str = ""
log_level: str = "INFO"
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
case_sensitive=False,
str_strip_whitespace=True,
)
@property
def trusted_redirect_origin_list(self) -> list[str]:
"""返回可信回跳来源列表"""
return _split_origins(self.trusted_redirect_origins)
@property
def cors_origin_list(self) -> list[str]:
"""返回允许跨域访问的来源列表"""
return _split_origins(self.cors_origins)
@lru_cache
def get_settings() -> Settings:
"""读取并缓存应用配置"""
return Settings()
app/core/database.py
+41
View File
@@ -0,0 +1,41 @@
# 管理异步数据库连接与会话
from collections.abc import AsyncIterator
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
AsyncAttrs,
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
from app.core.config import get_settings
class Base(AsyncAttrs, DeclarativeBase):
pass
settings = get_settings()
engine: AsyncEngine = create_async_engine(settings.database_url, pool_pre_ping=True)
async_session_factory = async_sessionmaker(engine, expire_on_commit=False)
async def get_db() -> AsyncIterator[AsyncSession]:
"""提供请求范围内的异步数据库会话"""
async with async_session_factory() as session:
yield session
async def check_database() -> None:
"""检查数据库连接是否可用"""
async with engine.connect() as connection:
await connection.execute(text("SELECT 1"))
async def close_database() -> None:
"""关闭数据库连接池"""
await engine.dispose()
app/core/exceptions.py
+66
View File
@@ -0,0 +1,66 @@
# 定义业务异常与全局异常处理
import logging
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from redis.exceptions import RedisError
from sqlalchemy.exc import SQLAlchemyError
from starlette.exceptions import HTTPException as StarletteHTTPException
from app.common.response import error_content
logger = logging.getLogger(__name__)
class BizException(Exception):
def __init__(self, code: int, msg: str = "请求失败") -> None:
"""初始化可安全返回给调用方的业务异常"""
self.code = code
self.msg = msg
super().__init__(msg)
async def handle_biz_exception(_request: Request, exc: BizException) -> JSONResponse:
"""将业务异常转换为统一失败响应"""
return JSONResponse(status_code=exc.code, content=error_content(exc.code, exc.msg))
async def handle_validation_exception(
_request: Request,
_exc: RequestValidationError,
) -> JSONResponse:
"""将请求校验异常转换为参数错误响应"""
return JSONResponse(status_code=400, content=error_content(400, "参数错误"))
async def handle_http_exception(_request: Request, exc: StarletteHTTPException) -> JSONResponse:
"""将框架 HTTP 异常转换为统一失败响应"""
code = exc.status_code if exc.status_code in {400, 401, 403, 429, 500} else 400
return JSONResponse(status_code=code, content=error_content(code))
async def handle_dependency_exception(
_request: Request,
exc: SQLAlchemyError | RedisError,
) -> JSONResponse:
"""将数据库和 Redis 故障转换为服务器异常响应"""
logger.error("外部依赖访问失败 error_type=%s", type(exc).__name__)
return JSONResponse(status_code=500, content=error_content(500))
async def handle_unexpected_exception(_request: Request, exc: Exception) -> JSONResponse:
"""将未捕获异常转换为服务器异常响应"""
logger.error("未处理的服务器异常 error_type=%s", type(exc).__name__)
return JSONResponse(status_code=500, content=error_content(500))
def register_exception_handlers(app: FastAPI) -> None:
"""注册应用级统一异常处理器"""
app.add_exception_handler(BizException, handle_biz_exception)
app.add_exception_handler(RequestValidationError, handle_validation_exception)
app.add_exception_handler(StarletteHTTPException, handle_http_exception)
app.add_exception_handler(SQLAlchemyError, handle_dependency_exception)
app.add_exception_handler(RedisError, handle_dependency_exception)
app.add_exception_handler(Exception, handle_unexpected_exception)
app/core/logging.py
+32
View File
@@ -0,0 +1,32 @@
# 配置日志与敏感信息脱敏
import logging
import re
SENSITIVE_REPLACEMENTS = (
(
re.compile(r"(?i)(password|token|authorization)(['\"\s:=]+)([^\s,}\]]+)"),
r"\1\2***",
),
(re.compile(r"(?i)(mysql\+asyncmy://[^:]+:)([^@]+)(@)"), r"\1***\3"),
)
class SensitiveDataFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
"""清理日志消息中的密码、Token 和连接凭据"""
message = record.getMessage()
for pattern, replacement in SENSITIVE_REPLACEMENTS:
message = pattern.sub(replacement, message)
record.msg = message
record.args = ()
return True
def configure_logging(level: str) -> None:
"""配置应用日志级别与敏感信息过滤器"""
root_logger = logging.getLogger()
root_logger.setLevel(level.upper())
sensitive_filter = SensitiveDataFilter()
for handler in root_logger.handlers:
handler.addFilter(sensitive_filter)
app/core/openapi.py
+59
View File
@@ -0,0 +1,59 @@
# 定制 OpenAPI 响应声明
from typing import Any
from fastapi import FastAPI
ERROR_RESPONSES: dict[str, tuple[int, ...]] = {
"/health": (500,),
"/api/auth/register": (400, 500),
"/api/auth/login": (400, 401, 500),
"/api/auth/verify": (401, 403, 500),
"/api/auth/logout": (401, 403, 500),
"/api/users/me": (401, 403, 500),
}
def configure_openapi(app: FastAPI) -> None:
"""配置只展示统一响应协议的 OpenAPI 文档"""
original_openapi = app.openapi
def custom_openapi() -> dict[str, Any]:
"""生成移除默认校验结构的 OpenAPI 文档"""
if app.openapi_schema is not None:
return app.openapi_schema
schema = original_openapi()
components = schema.setdefault("components", {}).setdefault("schemas", {})
components.pop("HTTPValidationError", None)
components.pop("ValidationError", None)
components["ApiErrorResponse"] = {
"type": "object",
"required": ["code", "data", "msg"],
"properties": {
"code": {"type": "integer", "example": 400},
"data": {"type": "null"},
"msg": {"type": "string", "example": "请求失败"},
},
}
for path, operations in schema.get("paths", {}).items():
for operation in operations.values():
if not isinstance(operation, dict) or "responses" not in operation:
continue
responses = operation["responses"]
responses.pop("422", None)
for code in ERROR_RESPONSES.get(path, (500,)):
responses[str(code)] = {
"description": "统一失败响应",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/ApiErrorResponse"}
}
},
}
app.openapi_schema = schema
return schema
app.openapi = custom_openapi
app/core/redis.py
+23
View File
@@ -0,0 +1,23 @@
# 管理共享 Redis 客户端
from redis.asyncio import Redis
from app.core.config import get_settings
settings = get_settings()
redis_client: Redis = Redis.from_url(settings.redis_url, decode_responses=True)
async def get_redis() -> Redis:
"""提供共享 Redis 客户端"""
return redis_client
async def check_redis() -> None:
"""检查 Redis 连接是否可用"""
await redis_client.ping()
async def close_redis() -> None:
"""关闭 Redis 客户端连接"""
await redis_client.aclose()
app/core/security.py
+52
View File
@@ -0,0 +1,52 @@
# 提供密码、Token 与回跳地址安全工具
import secrets
from urllib.parse import urlsplit
from pwdlib import PasswordHash
from app.core.exceptions import BizException
password_hash = PasswordHash.recommended()
def hash_password(password: str) -> str:
"""对用户密码执行安全哈希"""
return password_hash.hash(password)
def verify_password(password: str, encoded_password: str) -> bool:
"""验证明文密码与哈希是否匹配"""
try:
return password_hash.verify(password, encoded_password)
except (TypeError, ValueError):
return False
def generate_token() -> str:
"""生成不可预测的不透明 Token"""
return secrets.token_urlsafe(32)
def normalize_email(email: str) -> str:
"""规范化邮箱用于存储和比较"""
return email.strip().casefold()
def validate_from_url(from_url: str | None, trusted_origins: list[str]) -> str | None:
"""校验回跳地址是否属于可信来源"""
if from_url is None:
return None
value = from_url.strip()
parsed = urlsplit(value)
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
raise BizException(400, "参数错误")
if parsed.username is not None or parsed.password is not None:
raise BizException(400, "参数错误")
origin = f"{parsed.scheme.casefold()}://{parsed.netloc.casefold()}"
allowed_origins = {origin_value.casefold().rstrip("/") for origin_value in trusted_origins}
if origin not in allowed_origins:
raise BizException(400, "参数错误")
return value
app/main.py
+46
View File
@@ -0,0 +1,46 @@
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.router import api_router
from app.core.config import get_settings
from app.core.database import close_database
from app.core.exceptions import register_exception_handlers
from app.core.logging import configure_logging
from app.core.openapi import configure_openapi
from app.core.redis import close_redis
APP_TITLE = "Center Platform"
@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
"""管理应用启动与关闭生命周期"""
yield
await close_redis()
await close_database()
def create_app() -> FastAPI:
"""创建并配置 FastAPI 应用"""
settings = get_settings()
configure_logging(settings.log_level)
application = FastAPI(title=APP_TITLE, lifespan=lifespan)
register_exception_handlers(application)
application.include_router(api_router)
configure_openapi(application)
if settings.cors_origin_list:
application.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origin_list,
allow_credentials=False,
allow_methods=["*"],
allow_headers=["Authorization", "Content-Type"],
)
return application
app = create_app()
app/model/__init__.py
+3
View File
@@ -0,0 +1,3 @@
from app.model.user import User
__all__ = ["User"]
app/model/user.py
+47
View File
@@ -0,0 +1,47 @@
# 定义用户数据库模型
from datetime import datetime
from sqlalchemy import BigInteger, Boolean, DateTime, Integer, String, UniqueConstraint, func, text
from sqlalchemy.orm import Mapped, mapped_column
from app.core.database import Base
class User(Base):
__tablename__ = "users"
__table_args__ = (
UniqueConstraint("username", name="uq_users_username"),
UniqueConstraint("email", name="uq_users_email"),
{"comment": "中心平台用户表"},
)
id: Mapped[int] = mapped_column(
BigInteger().with_variant(Integer, "sqlite"),
primary_key=True,
autoincrement=True,
comment="用户唯一标识",
)
username: Mapped[str] = mapped_column(String(64), nullable=False, comment="登录用户名")
email: Mapped[str] = mapped_column(String(255), nullable=False, comment="用户邮箱")
password_hash: Mapped[str] = mapped_column(String(255), nullable=False, comment="密码哈希")
is_active: Mapped[bool] = mapped_column(
Boolean,
nullable=False,
default=True,
server_default=text("1"),
comment="用户是否启用",
)
created_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
comment="创建时间",
)
updated_at: Mapped[datetime] = mapped_column(
DateTime,
nullable=False,
server_default=func.now(),
onupdate=func.now(),
comment="更新时间",
)
app/repository/__init__.py
View File
app/repository/user_repository.py
+41
View File
@@ -0,0 +1,41 @@
# 封装用户数据访问操作
from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.model.user import User
class UserRepository:
def __init__(self, session: AsyncSession) -> None:
"""初始化用户数据访问对象"""
self.session = session
async def find_by_id(self, user_id: int) -> User | None:
"""根据用户标识查询用户"""
return await self.session.get(User, user_id)
async def find_by_username(self, username: str) -> User | None:
"""根据用户名进行不区分大小写查询"""
statement = select(User).where(func.lower(User.username) == username.casefold())
return await self.session.scalar(statement)
async def find_by_email(self, email: str) -> User | None:
"""根据规范化邮箱查询用户"""
statement = select(User).where(func.lower(User.email) == email.casefold())
return await self.session.scalar(statement)
async def find_by_account(self, account: str) -> User | None:
"""根据用户名或邮箱查询用户"""
normalized_account = account.strip().casefold()
statement = select(User).where(
or_(
func.lower(User.username) == normalized_account,
func.lower(User.email) == normalized_account,
)
)
return await self.session.scalar(statement)
def add(self, user: User) -> None:
"""将新用户加入数据库会话"""
self.session.add(user)
app/schema/__init__.py
View File
app/schema/auth.py
+16
View File
@@ -0,0 +1,16 @@
# 定义认证请求与响应模型
from pydantic import BaseModel, Field
class LoginRequest(BaseModel):
account: str = Field(min_length=1, max_length=255)
password: str = Field(min_length=1, max_length=128)
from_url: str | None = Field(default=None, max_length=2048)
class LoginResponse(BaseModel):
token: str
token_type: str = "Bearer"
expires_in: int
from_url: str | None = None
app/schema/user.py
+38
View File
@@ -0,0 +1,38 @@
# 定义用户请求与响应模型
from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
class UserRegisterRequest(BaseModel):
username: str = Field(min_length=3, max_length=64, pattern=r"^[A-Za-z0-9_.-]+$")
email: EmailStr = Field(max_length=255)
password: str = Field(min_length=8, max_length=128)
@field_validator("username")
@classmethod
def strip_username(cls, value: str) -> str:
"""清理用户名前后空白"""
return value.strip()
@field_validator("password")
@classmethod
def validate_password_strength(cls, value: str) -> str:
"""要求密码同时包含字母和数字"""
if not any(character.isalpha() for character in value):
raise ValueError("密码必须包含字母")
if not any(character.isdigit() for character in value):
raise ValueError("密码必须包含数字")
return value
class UserResponse(BaseModel):
id: int
username: str
email: EmailStr
is_active: bool
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
app/service/__init__.py
View File
app/service/auth_service.py
+63
View File
@@ -0,0 +1,63 @@
# 实现登录与身份认证业务
from redis.asyncio import Redis
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import Settings
from app.core.exceptions import BizException
from app.core.security import validate_from_url, verify_password
from app.model.user import User
from app.repository.user_repository import UserRepository
from app.schema.auth import LoginRequest, LoginResponse
from app.service.token_service import TokenService
class AuthService:
def __init__(
self,
session: AsyncSession,
redis_client: Redis,
settings: Settings,
) -> None:
"""初始化认证业务服务"""
self.users = UserRepository(session)
self.tokens = TokenService(
redis_client,
settings.token_expire_seconds,
settings.redis_key_prefix,
)
self.settings = settings
async def login(self, request: LoginRequest) -> LoginResponse:
"""验证账号密码并创建登录会话"""
from_url = validate_from_url(
request.from_url,
self.settings.trusted_redirect_origin_list,
)
user = await self.users.find_by_account(request.account)
if user is None or not verify_password(request.password, user.password_hash):
raise BizException(401, "账号或密码错误")
if not user.is_active:
raise BizException(401, "账号或密码错误")
token = await self.tokens.create(user.id)
return LoginResponse(
token=token,
expires_in=self.settings.token_expire_seconds,
from_url=from_url,
)
async def authenticate(self, token: str) -> User:
"""验证 Token 并返回当前有效用户"""
user_id = await self.tokens.get_user_id(token)
if user_id is None:
raise BizException(401, "登录已失效")
user = await self.users.find_by_id(user_id)
if user is None:
await self.tokens.delete(token)
raise BizException(401, "登录已失效")
if not user.is_active:
await self.tokens.delete(token)
raise BizException(403)
return user
app/service/token_service.py
+47
View File
@@ -0,0 +1,47 @@
# 管理 Redis Token 会话
from redis.asyncio import Redis
from app.core.security import generate_token
class TokenService:
def __init__(
self,
redis_client: Redis,
expire_seconds: int,
key_prefix: str,
) -> None:
"""初始化 Token 会话服务"""
self.redis = redis_client
self.expire_seconds = expire_seconds
self.key_prefix = key_prefix
def session_key(self, token: str) -> str:
"""构造单个 Token 的 Redis 键"""
return f"{self.key_prefix}:session:{token}"
async def create(self, user_id: int) -> str:
"""创建并保存新的用户 Token"""
token = generate_token()
await self.redis.set(
self.session_key(token),
str(user_id),
ex=self.expire_seconds,
)
return token
async def get_user_id(self, token: str) -> int | None:
"""读取 Token 对应的用户标识"""
value = await self.redis.get(self.session_key(token))
if value is None:
return None
try:
return int(value)
except (TypeError, ValueError):
await self.redis.delete(self.session_key(token))
return None
async def delete(self, token: str) -> None:
"""删除一个 Token"""
await self.redis.delete(self.session_key(token))
app/service/user_service.py
+40
View File
@@ -0,0 +1,40 @@
# 实现用户注册业务
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.exceptions import BizException
from app.core.security import hash_password, normalize_email
from app.model.user import User
from app.repository.user_repository import UserRepository
from app.schema.user import UserRegisterRequest
class UserService:
def __init__(self, session: AsyncSession) -> None:
"""初始化用户业务服务"""
self.session = session
self.repository = UserRepository(session)
async def register(self, request: UserRegisterRequest) -> User:
"""校验并创建中心平台用户"""
username = request.username.strip()
email = normalize_email(str(request.email))
if await self.repository.find_by_username(username):
raise BizException(400, "用户名已存在")
if await self.repository.find_by_email(email):
raise BizException(400, "邮箱已存在")
user = User(
username=username,
email=email,
password_hash=hash_password(request.password),
)
self.repository.add(user)
try:
await self.session.commit()
except IntegrityError as exc:
await self.session.rollback()
raise BizException(400, "请求失败") from exc
await self.session.refresh(user)
return user
migrations/env.py
+69
View File
@@ -0,0 +1,69 @@
import asyncio
from logging.config import fileConfig
from alembic import context
from sqlalchemy import Connection, pool
from sqlalchemy.ext.asyncio import async_engine_from_config
from app.core.config import get_settings
from app.core.database import Base
from app.model import User # noqa: F401
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
"""在不创建数据库连接时生成迁移 SQL"""
context.configure(
url=config.get_main_option("sqlalchemy.url"),
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
compare_type=True,
compare_server_default=True,
)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection: Connection) -> None:
"""在同步连接上下文中执行迁移"""
context.configure(
connection=connection,
target_metadata=target_metadata,
compare_type=True,
compare_server_default=True,
)
with context.begin_transaction():
context.run_migrations()
async def run_async_migrations() -> None:
"""创建异步连接并执行迁移"""
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
def run_migrations_online() -> None:
"""在在线模式下运行异步数据库迁移"""
asyncio.run(run_async_migrations())
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
migrations/script.py.mako
+26
View File
@@ -0,0 +1,26 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
revision: str = ${repr(up_revision)}
down_revision: str | Sequence[str] | None = ${repr(down_revision)}
branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
depends_on: str | Sequence[str] | None = ${repr(depends_on)}
def upgrade() -> None:
"""升级数据库结构"""
${upgrades if upgrades else "pass"}
def downgrade() -> None:
"""回滚数据库结构"""
${downgrades if downgrades else "pass"}
migrations/versions/20260904_0001_create_users.py
+63
View File
@@ -0,0 +1,63 @@
"""创建中心平台用户表
Revision ID: 20260904_0001
Revises:
Create Date: 2026-09-04
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20260904_0001"
down_revision: str | Sequence[str] | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""创建带中文注释的用户表"""
op.create_table(
"users",
sa.Column(
"id",
sa.BigInteger(),
autoincrement=True,
nullable=False,
comment="用户唯一标识",
),
sa.Column("username", sa.String(length=64), nullable=False, comment="登录用户名"),
sa.Column("email", sa.String(length=255), nullable=False, comment="用户邮箱"),
sa.Column("password_hash", sa.String(length=255), nullable=False, comment="密码哈希"),
sa.Column(
"is_active",
sa.Boolean(),
server_default=sa.text("1"),
nullable=False,
comment="用户是否启用",
),
sa.Column(
"created_at",
sa.DateTime(),
server_default=sa.text("CURRENT_TIMESTAMP"),
nullable=False,
comment="创建时间",
),
sa.Column(
"updated_at",
sa.DateTime(),
server_default=sa.text("CURRENT_TIMESTAMP"),
nullable=False,
comment="更新时间",
),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("email", name="uq_users_email"),
sa.UniqueConstraint("username", name="uq_users_username"),
comment="中心平台用户表",
)
def downgrade() -> None:
"""删除中心平台用户表"""
op.drop_table("users")
pyproject.toml
+40
View File
@@ -0,0 +1,40 @@
[project]
name = "center"
version = "0.1.0"
description = "轻量级统一认证中心"
requires-python = ">=3.11"
dependencies = [
"alembic>=1.16.0",
"asyncmy>=0.2.10",
"email-validator>=2.2.0",
"fastapi[standard]>=0.141.1",
"pwdlib[argon2]>=0.3.0",
"pydantic-settings>=2.10.0",
"redis>=6.2.0",
"sqlalchemy[asyncio]>=2.0.41",
]
[dependency-groups]
dev = [
"aiosqlite>=0.21.0",
"fakeredis>=2.30.0",
"httpx>=0.28.1",
"pytest>=8.4.0",
"pytest-asyncio>=1.0.0",
"ruff>=0.12.0",
]
[tool.pytest.ini_options]
asyncio_mode = "auto"
pythonpath = ["."]
testpaths = ["tests"]
[tool.ruff]
line-length = 100
target-version = "py311"
[tool.ruff.lint]
select = ["ASYNC", "B", "E", "F", "I", "UP"]
[tool.ruff.format]
quote-style = "double"
uv.lock
Generated
+1862
View File
File diff suppressed because it is too large Load Diff