ORM 应用

1. 生成模板

pip install sqlacodegen

pip install pymysql

sqlacodegen mysql+pymysql://username:password@127.0.0.1:3306/db_name > models_tmp.py

2. 操作实例

data = session.query(Emp).from_statement(text('select * from emp where id=2')).params(name='123').all()

3. 生成单表

sqlacodegen --tables third_api "mysql+pymysql://username:password@host/db_name" >tmp.py

4. 基础查询

等于查询：

query.filter(User.name == 'ed')

不等于查询：

query.filter(User.name != 'ed')

like查询：

query.filter(User.name.like('%ed%'))

in查询：

query.filter(User.name.in_(['ed', 'wendy', 'jack']))

not in 查询

query.filter(~User.name.in_(['ed', 'wendy', 'jack']))

is null查询：

query.filter(User.name == None)

is not null查询

query.filter(User.name != None)

and查询：

from sqlalchemy import and_

query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))

# or send multiple expressions to .filter()
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')


or查询：

from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))

match查询

query.filter(User.name.match('wendy'))

order_by语句：

query = session.query(User).filter(User.name.like('%ed')).order_by(User.id)

count计数

session.query(User).filter(User.name.like('%ed')).count()

分页查询

session.query(Channel).filter(and_(
                filter_list
            )).order_by(desc(Channel.updateTime)).slice(offset, limit)

5. oracle 实例

# orm_oracle_example.py

import oracledb
oracledb.init_oracle_client(lib_dir="D:\\tool\\instantclient_23_9")
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Sequence
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime



# ==================== 配置部分 ====================

# 如果使用旧版 Oracle (如 12c, 18c)，请取消注释下面这行并安装 Instant Client
# oracledb.init_oracle_client(mode=oracledb.CLIENT_THICK)

# 替换为你的数据库连接信息
DATABASE_URL = "oracle+oracledb://learun:lhadmin@qixin.bhgmarketplace.com:1521/qixindb"

# ==================== ORM 配置 ====================

engine = create_engine(DATABASE_URL, echo=True)  # echo=True 可查看 SQL
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# ==================== 定义模型 ====================

class User(Base):
    __tablename__ = 'users11'

    # Oracle 不支持自增，使用 Sequence
    id = Column(Integer, Sequence('user_id_seq'), primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

# ==================== 主程序 ====================

def main():
    print("正在连接数据库...")
    # 创建表（如果不存在）
    Base.metadata.create_all(bind=engine)
    # print("表已创建或已存在。")

    # 获取数据库会话
    db = SessionLocal()

    try:
        # 1. 创建 (Create)
        print("\n--- 创建用户 ---")
        new_user = User(name="张三", email="zhangsan@example.com")
        db.add(new_user)
        db.commit()
        db.refresh(new_user)  # 刷新以获取数据库生成的 ID
        print(f"创建成功: {new_user}")

        # 2. 查询 (Read)
        print("\n--- 查询用户 ---")
        user = db.query(User).filter(User.email == "zhangsan@example.com").first()
        print(f"查询结果: {user}")

        # 3. 更新 (Update)
        print("\n--- 更新用户 ---")
        if user:
            user.name = "李四"
            db.commit()
            print(f"更新成功: {user}")

        # 4. 删除 (Delete)
        print("\n--- 删除用户 ---")
        if user:
            db.delete(user)
            db.commit()
            print("删除成功")

    except Exception as e:
        print(f"发生错误: {e}")
        db.rollback()
    finally:
        db.close()
        print("\n数据库会话已关闭。")

if __name__ == "__main__":
    main()