Python ORM 框架：SQLAlchemy 完全指南 | 极客日志

Python ORM 框架：SQLAlchemy 完全指南 | 极客日志

pip install sqlalchemy

from sqlalchemy import create_engine # 创建引擎 engine = create_engine('sqlite:///example.db') # MySQL 示例：create_engine('mysql+pymysql://user:password@host:port/dbname')

from sqlalchemy.orm import sessionmaker Session = sessionmaker(bind=engine) session = Session()

from sqlalchemy import create_engine, Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime # 创建基类 Base = declarative_base() # 定义用户模型 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(50), unique=True, nullable=False) email = Column(String(100), unique=True, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) def __repr__(self): return f"<User(username='{self.username}', email='{self.email}')>" # 创建数据库引擎 engine = create_engine('sqlite:///example.db', echo=True) # 创建表 Base.metadata.create_all(engine) # 创建会话 Session = sessionmaker(bind=engine) session = Session() # 创建用户 new_user = User(username='alice', email='[email protected]') session.add(new_user) session.commit() # 查询用户 users = session.query(User).all() for user in users: print(user) # 关闭会话 session.close()

# 单条插入 user = User(username='bob', email='[email protected]') session.add(user) session.commit() # 批量插入 users = [ User(username='user1', email='[email protected]'), User(username='user2', email='[email protected]') ] session.add_all(users) session.commit()

# 基本查询 users = session.query(User).all() first_user = session.query(User).first() user_by_id = session.query(User).get(1) # 条件查询 active_users = session.query(User).filter(User.is_active == True).all() user_by_email = session.query(User).filter(User.email == '[email protected]').first() # 复杂查询 from sqlalchemy import and_, or_, not_ # AND 条件 users = session.query(User).filter( and_(User.is_active == True, User.username.like('%alice%')) ).all() # OR 条件 users = session.query(User).filter( or_(User.username.like('%alice%'), User.email.like('%example%')) ).all() # 排序和限制 recent_users = session.query(User).order_by(User.created_at.desc()).limit(10).all() # 分页查询 page = 1 per_page = 20 users = session.query(User).offset((page-1)*per_page).limit(per_page).all()

# 更新单条记录 user = session.query(User).filter(User.id == 1).first() if user: user.email = '[email protected]' session.commit() # 批量更新 session.query(User).filter(User.is_active == False).update({ User.is_active: True }) session.commit()

# 删除单条记录 user = session.query(User).filter(User.id == 1).first() if user: session.delete(user) session.commit() # 批量删除 session.query(User).filter(User.is_active == False).delete() session.commit()

from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime # 创建数据库连接 engine = create_engine('sqlite:///example.db', echo=True) Base = declarative_base() # 定义模型 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(50), unique=True, nullable=False) email = Column(String(100), unique=True, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) # 创建表 Base.metadata.create_all(engine) # 创建会话工厂 Session = sessionmaker(bind=engine) # 创建会话 session = Session() try: # 开始事务 user = User(username='transaction_user', email='[email protected]') session.add(user) # 提交事务 session.commit() print("Transaction completed successfully") except Exception as e: # 回滚事务 session.rollback() print(f"Transaction failed: {e}") finally: session.close()

from contextlib import contextmanager from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime # 创建数据库连接 engine = create_engine('sqlite:///example.db', echo=True) Base = declarative_base() # 定义模型 class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, autoincrement=True) username = Column(String(50), unique=True, nullable=False) email = Column(String(100), unique=True, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) # 创建表 Base.metadata.create_all(engine) # 创建会话工厂 Session = sessionmaker(bind=engine) # 创建会话 session = Session() @contextmanager def get_db_session(): session = Session() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close() # 使用示例 with get_db_session() as session: user = User(username='context_user', email='[email protected]') session.add(user)

from sqlalchemy import create_engine engine = create_engine( 'mysql+pymysql://user:password@localhost/dbname', pool_size=10, # 连接池大小 max_overflow=20, # 最大溢出连接数 pool_timeout=30, # 连接超时时间 pool_recycle=3600 # 连接回收时间 )