kepler.atlas 0.1.6

Easy-to-use ORM for database operations with lazy table reflection

kepler-atlas

基于 SQLAlchemy 的 Python ORM 库，专为交互式编程（Jupyter/IPython）优化，提供更简洁的 API。

特性

• 🚀 懒加载表反射 - 按需加载表结构，启动速度快
• 🔤 大小写不敏感 - db.users、db.USERS、db.Users 都能访问同一张表
• 📊 DataFrame 集成 - 查询结果一键转为 pandas DataFrame
• 🗄️ 多数据库支持 - MySQL、PostgreSQL、SQLite、Oracle、MSSQL 等
• ✨ Tab 补全 - IPython/Jupyter 中自动提示表名

安装

pip install kepler-atlas

快速开始

连接数据库

from kepler.atlas import DataBase

# 方式1: 连接字符串
db = DataBase('mysql://user:pass@localhost/db')

# 方式2: SQLAlchemy Engine
from sqlalchemy import create_engine
engine = create_engine('postgresql://user:pass@localhost/db')
db = DataBase(engine)

# 方式3: 上下文管理器（自动关闭连接）
with DataBase('sqlite:///example.db') as db:
    ...

访问表

# 大小写不敏感
users = db.users
users = db.USERS
users = db.Users  # 都指向同一张表

# 列出所有表
print(db.tables)

查询

# 基本查询
db.query(db.users).all()

# 条件过滤
db.query(db.users).filter(db.users.age > 18).all()

# 返回 DataFrame
df = db.query(db.users).filter(db.users.age > 18).to_df()

# 排序、分页
db.query(db.users).order_by(db.users.id.desc()).limit(10).all()

插入

# 单条插入
user = db.users(name='Alice', email='[e-mail]')
db.update(user)
db.commit()

# 批量插入 - DataFrame
import pandas as pd
df = pd.DataFrame({
    'name': ['Alice', 'Bob'],
    'email': ['[e-mail]', '[e-mail]']
})
db.insert(db.users, df)
db.commit()

# 批量插入 - 字典列表
data = [{'name': 'Alice'}, {'name': 'Bob'}]
db.insert(db.users, data, ignore=True)  # ignore=True 忽略重复键
db.commit()

更新

# 查询后修改
user = db.query(db.users).filter(db.users.name == 'Alice').first()
user.email = '[e-mail]'
db.commit()

# 批量更新
db.query(db.users).filter(db.users.age < 18).update({'status': 'minor'})
db.commit()

# merge（存在则更新，不存在则插入）
user = db.users(name='Alice', email='[e-mail]')
db.merge(user)
db.commit()

删除

# 删除单条
user = db.query(db.users).filter(db.users.name == 'Alice').first()
db.delete(user)
db.commit()

# 批量删除
db.query(db.users).filter(db.users.status == 'inactive').delete()
db.commit()

自定义模型

from sqlalchemy import Column, Integer, String

class User(db.Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

db.create_all()  # 创建表

API 参考

DataBase

方法/属性	说明
DataBase(bind, schema=None)	创建数据库实例
db.tables	返回所有表名列表
db.Base	获取 automap 基类
db.query(*entities)	创建查询
db.insert(table, data, ignore=True)	批量插入
db.update(obj)	添加对象到会话
db.merge(obj)	合并对象
db.delete(obj)	删除对象
db.commit()	提交事务
db.rollback()	回滚事务
db.flush()	刷新到数据库（不提交）
db.close()	关闭连接
db.refresh()	刷新元数据

Query

方法	说明
.filter(condition)	过滤条件
.order_by(column)	排序
.limit(n)	限制数量
.offset(n)	偏移量
.all()	返回所有结果
.first()	返回第一条
.one()	返回唯一一条
.count()	计数
.to_df()	转为 DataFrame

异常

from kepler.atlas import (
    AtlasError,       # 基础异常
    TableError,       # 表操作错误
    ConnectionError,  # 连接错误
    InsertError,      # 插入错误
    ValidationError,  # 验证错误
)

交互式编程

在 IPython/Jupyter 中，Tab 补全会自动显示表名：

db = DataBase('sqlite:///example.db')
db.u<Tab>  # 显示: db.users

列名也同样支持：

数据库方言支持

数据库	忽略重复键方式
MySQL	INSERT IGNORE
SQLite	INSERT OR REPLACE
PostgreSQL	ON CONFLICT DO NOTHING
Oracle	/*+ IGNORE_ROW_ON_DUPKEY_INDEX */
MSSQL	IGNORE_DUP_KEY

注意事项

• ⚠️ 表必须有主键才能被 ORM 映射

License

GPL-3.0