datamanager_hjy

一句话宣言: 通用的数据管理脚手架,支持多数据库、配置驱动、高性能,让数据操作变得毫不费力。
🎯 优雅的"Hello, World"
from datamanager_hjy import DataManager
dm = DataManager()
result = dm.create('users', {'name': '张三', 'email': 'zhangsan@example.com'})
users = dm.query('users').filter(name='张三').all()
print(f"🎉 数据操作完成: {users}")
✨ 为什么选择 datamanager_hjy?
🚀 零摩擦体验
- 一键安装:
pip install datamanager-hjy
- 零配置启动: 自动检测环境,智能配置
- 所见即所得: README中的示例代码可直接运行
🎨 苹果产品级设计
- 高内聚低耦合: 内部复杂,接口简洁
- 类型安全: 完整的类型提示,IDE友好
- 优雅错误处理: 人类可读的错误信息
🌟 云原生公民
- 配置即插即用: 外置配置,依赖注入
- 生命周期管理: 完整的启动、运行、关闭控制
- 健康检查: 实时监控,自动恢复
🔄 多数据库支持
- 统一接口: 一套API操作多种数据库
- 连接池管理: 智能连接池,支持动态扩缩容
- 读写分离: 支持主从数据库配置
- 故障转移: 自动故障检测和切换
🚀 快速开始
安装
pip install datamanager-hjy
快速开始
python examples/quick_start.py
datamanager --demo
基本使用
from datamanager_hjy import DataManager
dm = DataManager()
user = dm.create('users', {
'name': '李四',
'email': 'lisi@example.com',
'age': 25
})
users = dm.query('users').filter(age__gte=20).all()
dm.update('users', {'age': 26}, {'id': user['id']})
dm.delete('users', {'id': user['id']})
高级功能
users_data = [
{'name': '王五', 'email': 'wangwu@example.com'},
{'name': '赵六', 'email': 'zhaoliu@example.com'}
]
dm.batch_create('users', users_data)
with dm.transaction() as txn:
txn.create('users', user_data)
txn.update('profiles', profile_data, condition)
txn.commit()
with dm.multi_database_transaction(['default', 'analytics']) as txn:
txn.create('users', user_data, database='default')
txn.create('analytics_data', analytics_data, database='analytics')
config = dm.get_config('database.default')
dm.update_config('database.default', new_config)
dm.reload_config()
配置即插即用
databases:
default:
type: mysql
host: ${DB_HOST}
port: ${DB_PORT}
database: ${DB_NAME}
username: ${DB_USER}
password: ${DB_PASSWORD}
pool_size: 10
max_overflow: 20
read_replica:
type: mysql
host: ${DB_READ_HOST}
port: ${DB_READ_PORT}
database: ${DB_READ_NAME}
username: ${DB_READ_USER}
password: ${DB_READ_PASSWORD}
pool_size: 5
max_overflow: 10
models:
auto_discover: true
base_path: models
cache_enabled: true
cache_ttl: 3600
monitoring:
enabled: true
metrics_collection: true
slow_query_threshold: 1000
connection_monitoring: true
📊 性能指标
- 连接建立时间: < 100ms
- 查询响应时间: < 50ms (简单查询)
- 批量操作: 1000条记录 < 1秒
- 连接池效率: 连接复用率 > 90%
- 内存使用: 连接池内存 < 100MB
- 并发支持: 1000+ 并发连接
- 多数据库: MySQL、PostgreSQL、SQLite
🎯 支持的功能
数据库支持
- MySQL: 完整支持,包括连接池、读写分离
- PostgreSQL: 完整支持,包括JSON字段、数组类型
- SQLite: 本地开发支持,轻量级应用
核心功能
- CRUD操作: 完整的增删改查接口
- 批量操作: 高效的批量插入、更新、删除
- 事务管理: 自动事务管理和回滚
- 查询优化: 智能查询优化和索引建议
- 连接池: 智能连接池,支持动态扩缩容
- 读写分离: 支持主从数据库配置
- 故障转移: 自动故障检测和切换
高级功能
- 配置管理: 配置存储在数据库,支持热更新
- 模型管理: 自动发现和注册数据模型
- 监控系统: 性能监控、慢查询检测
- 缓存集成: 与Redis缓存系统集成
- 日志系统: 完整的操作日志和错误日志
🔧 扩展功能
自定义适配器
from datamanager_hjy.adapters.base import BaseAdapter
class CustomAdapter(BaseAdapter):
def connect(self):
pass
def execute(self, query, params=None):
pass
自定义模型
from datamanager_hjy.models.base import BaseModel
class UserModel(BaseModel):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False)
email = Column(String(255), unique=True)
def validate_email(self, email):
return '@' in email
📚 文档
🧪 测试
pytest
pytest tests/unit/
pytest tests/integration/
pytest tests/performance/ -m "slow"
pytest --cov=datamanager_hjy tests/
📈 监控
性能指标
- 连接池状态: 连接数量、使用率、等待时间
- 查询性能: 查询时间、慢查询数量
- 错误率: 连接错误、查询错误
- 资源使用: CPU、内存、磁盘使用
监控接口
metrics = dm.get_metrics()
status = dm.get_connection_status()
slow_queries = dm.get_slow_queries()
🤝 贡献
我们欢迎所有形式的贡献!
贡献方式
开发环境
git clone https://github.com/hjy/datamanager_hjy.git
cd datamanager_hjy
pip install -e ".[dev]"
pytest
black datamanager_hjy/
isort datamanager_hjy/
mypy datamanager_hjy/
📄 许可证
本项目采用 MIT 许可证 - 查看 LICENSE 文件了解详情。
🆘 支持
datamanager_hjy - 让数据操作变得毫不费力 ✨