xtsqlorm 0.1.0

Add your description here

xtsqlorm

现代化的 Python SQLAlchemy ORM 框架

English | 简体中文

---

✨ 特性

🏗️ 现代化架构

• ✅ 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

---

🚀 快速开始

1. 配置数据库

# 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,
    }
}

2. 定义模型

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)

3. 基础 CRUD 操作

from xtsqlorm import create_repository

# 创建仓储
user_repo = create_repository(User, db_key='default')

# 创建
user = user_repo.create({
    'username': 'john_doe',
    'email': '[e-mail]',
    'phone': '13800138000'
})

# 查询
user = user_repo.get_by_id(1)
all_users = user_repo.get_all(limit=10)

# 更新
updated_user = user_repo.update(1, {'email': '[e-mail]'})

# 删除
user_repo.delete(1)

4. 高级功能

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': '[e-mail]'},
    {'username': 'user2', 'email': '[e-mail]'},
])

# 字段统计
stats = ops.get_field_stats('created_at')

# 导出数据
df = ops.export_to_dataframe(limit=100)

5. 异步操作

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': '[e-mail]'
    })

asyncio.run(main())

---

📚 核心组件

Repository (仓储模式)

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()

Mixins (模型扩展)

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)              │
└─────────────────────────────────────────────┘

核心原则

• 依赖倒置: 依赖抽象接口而非具体实现
• 单一职责: 每个组件只负责一个功能
• 开闭原则: 对扩展开放，对修改封闭
• 组合优于继承: 使用组合模式构建功能

---

📖 文档

• 完整 API 文档
• 架构设计文档
• 示例代码
• 变更日志
• 贡献指南

---

🧪 示例代码

项目包含 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/

---

🤝 贡献

我们欢迎所有形式的贡献！请阅读 贡献指南 了解详情。

快速开始贡献

1. Fork 本仓库
2. 创建特性分支: git checkout -b feature/amazing-feature
3. 提交更改: git commit -m 'Add amazing feature'
4. 推送分支: git push origin feature/amazing-feature
5. 提交 Pull Request

---

📝 变更日志

详见 CHANGELOG.md

最新版本 v0.1.0 (2025-10-25)

新增:

• ✨ 完整的异步架构实现
• ✨ 代码质量优化 (Ruff + basedPyright)
• ✨ 增强的数据验证器

修复:

• 🐛 异步资源泄漏问题
• 🐛 类型检查错误
• 🐛 全角标点符号规范

---

📄 许可证

本项目采用 MIT License 开源协议。

---

💖 致谢

感谢以下开源项目：

• SQLAlchemy - 强大的 Python SQL 工具包
• Pydantic - 数据验证
• Ruff - 代码质量工具
• basedPyright - 类型检查

---

📮 联系方式

• 作者: sandorn
• 邮箱: [e-mail]
• 仓库: Gitee | GitHub

---

如果这个项目对你有帮助，请给一个 ⭐ Star！

Made with ❤️ by sandorn