- ✅ SOLID 原则: 接口清晰，职责分明
- ✅ 依赖注入: ConnectionManager → SessionProvider → Repository
- ✅ 异步支持: 完整的异步架构实现
- ✅ 类型安全: 100% 类型注解覆盖
- ✅ 连接池管理: 自动连接池优化
- ✅ 批量操作:
bulk_create,bulk_update高效处理 - ✅ 查询缓存: 内置 LRU 缓存支持
- ✅ 分页查询: 优化的分页实现
- ✅ 数据验证: 集成 Pydantic 验证
- ✅ 软删除: 内置软删除支持
- ✅ 乐观锁: 版本控制防止并发冲突
- ✅ 事务管理: 自动事务回滚
- ✅ Repository 模式: 简化 CRUD 操作
- ✅ Mixin 扩展: 时间戳、软删除、版本控制
- ✅ 表反射: 动态映射数据库表
- ✅ 丰富示例: 8+ 完整示例代码
pip install xtsqlorm# 克隆仓库
git clone https://gitee.com/sandorn/xtsqlorm.git
cd xtsqlorm
# 安装依赖 (使用 uv)
uv sync
# 或使用 pip
pip install -e ".[dev]"# 异步支持 (MySQL)
pip install aiomysql
# 数据验证
pip install pydantic
# 数据导出
pip install pandas
# 日志增强
pip install xtlog# xtsqlorm/cfg.py
DB_Config = {
'default': {
'driver': 'mysql+pymysql',
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'your_password',
'database': 'your_database',
'charset': 'utf8mb4',
'pool_size': 5,
'max_overflow': 10,
}
}from xtsqlorm import BaseModel, IdMixin, TimestampMixin
from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import String
class User(BaseModel, IdMixin, TimestampMixin):
__tablename__ = 'users'
username: Mapped[str] = mapped_column(String(50), unique=True)
email: Mapped[str] = mapped_column(String(100))
phone: Mapped[str] = mapped_column(String(20), nullable=True)from xtsqlorm import create_repository
# 创建仓储
user_repo = create_repository(User, db_key='default')
# 创建
user = user_repo.create({
'username': 'john_doe',
'email': 'john@example.com',
'phone': '13800138000'
})
# 查询
user = user_repo.get_by_id(1)
all_users = user_repo.get_all(limit=10)
# 更新
updated_user = user_repo.update(1, {'email': 'new@example.com'})
# 删除
user_repo.delete(1)from xtsqlorm import create_orm_operations
# 创建 ORM 操作对象 (带验证和缓存)
ops = create_orm_operations(User, db_key='default', cache_enabled=True)
# 分页查询
users, total = ops.get_paginated(page=1, per_page=10)
# 批量创建
users = ops.bulk_create([
{'username': 'user1', 'email': 'user1@example.com'},
{'username': 'user2', 'email': 'user2@example.com'},
])
# 字段统计
stats = ops.get_field_stats('created_at')
# 导出数据
df = ops.export_to_dataframe(limit=100)import asyncio
from xtsqlorm import create_async_repository
async def main():
# 创建异步仓储
async_repo = create_async_repository(User, db_key='default')
# 异步查询
users = await async_repo.get_all(limit=10)
user = await async_repo.get_by_id(1)
# 异步创建
new_user = await async_repo.create({
'username': 'async_user',
'email': 'async@example.com'
})
asyncio.run(main())from xtsqlorm import Repository, IRepository
class UserRepository(Repository[User]):
"""用户仓储 - 标准 CRUD 操作"""
def find_by_username(self, username: str) -> User | None:
"""自定义查询方法"""
with self._session_provider.transaction() as session:
return session.query(self._model).filter_by(username=username).first()from xtsqlorm import (
IdMixin, # 主键 ID
TimestampMixin, # 创建/更新时间
SoftDeleteMixin, # 软删除
VersionedMixin, # 乐观锁版本控制
UTCTimeMixin, # UTC 时间戳
)
class Article(BaseModel, IdMixin, TimestampMixin, SoftDeleteMixin, VersionedMixin):
__tablename__ = 'articles'
title: Mapped[str] = mapped_column(String(200))
content: Mapped[str] = mapped_column(Text)from xtsqlorm import JsonEncodedDict, EnumType, UTCDateTime
class Config(BaseModel):
__tablename__ = 'configs'
# JSON 字段
settings: Mapped[dict] = mapped_column(JsonEncodedDict)
# 枚举字段
status: Mapped[str] = mapped_column(EnumType('active', 'inactive'))
# UTC 时间字段
expires_at: Mapped[datetime] = mapped_column(UTCDateTime)from pydantic import BaseModel as PydanticModel, field_validator
from xtsqlorm import create_orm_operations, validate_email, validate_phone
class UserValidator(PydanticModel):
username: str
email: str
phone: str
@field_validator('email')
@classmethod
def validate_email_format(cls, v: str) -> str:
return validate_email(v, 'email')
@field_validator('phone')
@classmethod
def validate_phone_format(cls, v: str) -> str:
return validate_phone(v, 'phone')
# 使用验证
ops = create_orm_operations(User, db_key='default', validator_model=UserValidator)┌─────────────────────────────────────────────┐
│ Application Layer │
│ (Controller / Service / Business Logic) │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ ORM Operations Layer │
│ (OrmOperations / AsyncOrmOperations) │
│ - Data Validation │
│ - Query Caching │
│ - Advanced Features │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Repository Layer │
│ (Repository / AsyncRepository) │
│ - CRUD Operations │
│ - Transaction Management │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Session Provider Layer │
│ (SessionProvider / AsyncSessionProvider) │
│ - Session Lifecycle │
│ - Transaction Context │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Connection Manager Layer │
│ (ConnectionManager / AsyncConnectionManager)│
│ - Connection Pool │
│ - Engine Management │
└─────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────┐
│ Database Layer │
│ (MySQL / PostgreSQL / SQLite) │
└─────────────────────────────────────────────┘
- 依赖倒置: 依赖抽象接口而非具体实现
- 单一职责: 每个组件只负责一个功能
- 开闭原则: 对扩展开放，对修改封闭
- 组合优于继承: 使用组合模式构建功能
项目包含 8+ 完整示例，涵盖所有核心功能：
| 示例文件 | 功能说明 |
|---|---|
example_01_basic_sync.py |
基础同步 CRUD 操作 |
example_02_advanced_operations.py |
高级查询和批量操作 |
example_03_table_reflection.py |
表反射和动态模型 |
example_04_mixins_and_types.py |
Mixin 和自定义类型 |
example_05_data_validation.py |
数据验证和 Pydantic |
example_06_transactions.py |
事务管理和 UnitOfWork |
example_07_complete_workflow.py |
完整用户管理流程 |
example_08_table_management.py |
表管理工具 |
examples_async.py |
异步操作完整示例 |
运行示例：
# 使用 uv (推荐)
uv run python examples/example_01_basic_sync.py
# 或使用 python
python examples/example_01_basic_sync.py# 运行所有测试
pytest
# 查看测试覆盖率
pytest --cov=xtsqlorm --cov-report=html
# 运行类型检查
basedpyright xtsqlorm/
# 代码质量检查
ruff check xtsqlorm/ examples/我们欢迎所有形式的贡献！请阅读 贡献指南 了解详情。
- Fork 本仓库
- 创建特性分支:
git checkout -b feature/amazing-feature - 提交更改:
git commit -m 'Add amazing feature' - 推送分支:
git push origin feature/amazing-feature - 提交 Pull Request
详见 CHANGELOG.md
新增:
- ✨ 完整的异步架构实现
- ✨ 代码质量优化 (Ruff + basedPyright)
- ✨ 增强的数据验证器
修复:
- 🐛 异步资源泄漏问题
- 🐛 类型检查错误
- 🐛 全角标点符号规范
本项目采用 MIT License 开源协议。
感谢以下开源项目：
- SQLAlchemy - 强大的 Python SQL 工具包
- Pydantic - 数据验证
- Ruff - 代码质量工具
- basedPyright - 类型检查
- 作者: sandorn
- 邮箱: sandorn@live.cn
- 仓库: Gitee | GitHub
如果这个项目对你有帮助，请给一个 ⭐ Star！
Made with ❤️ by sandorn