SQLAlchemy 核心用法与实战案例解析

SQLAlchemy 核心用法与实战案例解析 | 极客日志

# 安装核心库
pip install sqlalchemy
# 安装数据库驱动（根据实际使用选择）
pip install psycopg2-binary # PostgreSQL
pip install pymysql # MySQL
pip install cx-Oracle # Oracle
pip install pyodbc # SQL Server
# 如果需要数据库迁移工具
pip install alembic

import sqlalchemy
print(f"SQLAlchemy 版本：{sqlalchemy.__version__}")

from sqlalchemy import create_engine, text

# 创建 SQLite 内存数据库（开发测试）
engine = create_engine('sqlite:///:memory:', echo=True)

# 创建文件型 SQLite 数据库
engine = create_engine('sqlite:///./myapp.db', echo=False)

# 连接 PostgreSQL
# engine = create_engine('postgresql://user:password@localhost/mydb')

# 连接 MySQL
# engine = create_engine('mysql+pymysql://user:password@localhost/mydb')

# 测试连接
with engine.connect() as conn:
    result = conn.execute(text("SELECT 1"))
    print(result.fetchone())

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, Boolean, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, 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, index=True)
    email = Column(String(120), unique=True, nullable=False)
    password_hash = Column(String(128), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 关联关系
    posts = relationship("Post", back_populates="author", cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"<User(id={self.id}, username={self.username})>"

# 定义文章表
class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    views = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # 外键
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    
    # 反向关系
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")
    
    def __repr__(self):
        return f"<Post(id={self.id}, title={self.title[:20]})>"

# 定义评论表
class Comment(Base):
    __tablename__ = 'comments'
    id = Column(Integer, primary_key=True, autoincrement=True)
    content = Column(Text, nullable=False)
    author_name = Column(String(50))
    created_at = Column(DateTime, default=datetime.utcnow)
    post_id = Column(Integer, ForeignKey('posts.id'), nullable=False)
    post = relationship("Post", back_populates="comments")

# 创建所有表
engine = create_engine('sqlite:///./blog.db', echo=True)
Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker

# 创建会话工厂
SessionLocal = sessionmaker(bind=engine)

# 创建会话
session = SessionLocal()

# ---------- 创建数据 ----------
# 创建用户
user1 = User(username='张三', email='[email protected]', password_hash='hashed_password_123')
user2 = User(username='李四', email='[email protected]', password_hash='hashed_password_456')

# 批量添加
session.add(user1)
session.add(user2)
session.commit()  # 提交事务

# 创建文章
post1 = Post(
    title='SQLAlchemy 入门指南',
    content='本文详细介绍 SQLAlchemy 的基本使用方法...',
    author=user1
)
post2 = Post(
    title='Python 数据库编程最佳实践',
    content='在实际项目中如何高效使用 ORM...',
    author=user1
)
session.add_all([post1, post2])
session.commit()

# ---------- 查询数据 ----------
# 查询所有用户
users = session.query(User).all()
print(f"用户数量：{len(users)}")

# 带条件的查询
user = session.query(User).filter(User.username == '张三').first()
print(f"找到用户：{user}")

# 模糊查询
posts = session.query(Post).filter(Post.title.contains('SQLAlchemy')).all()

# 关联查询
user_with_posts = session.query(User).options(joinedload(User.posts)).filter(User.id == 1).first()
print(f"用户 {user_with_posts.username} 的文章数量：{len(user_with_posts.posts)}")

# 聚合查询
from sqlalchemy import func
post_count = session.query(func.count(Post.id)).scalar()
print(f"文章总数：{post_count}")

# ---------- 更新数据 ----------
post = session.query(Post).first()
post.views += 1
post.title = "更新后的标题"
session.commit()

# ---------- 删除数据 ----------
# 删除单条记录
comment_to_delete = session.query(Comment).filter(Comment.id == 1).first()
if comment_to_delete:
    session.delete(comment_to_delete)
session.commit()

# 关闭会话
session.close()

from contextlib import contextmanager

@contextmanager
def session_scope():
    """提供会话的上下文管理器"""
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# 使用示例
with session_scope() as session:
    users = session.query(User).filter(User.is_active == True).all()
    for user in users:
        print(user.username)

from sqlalchemy import and_, or_, not_, desc, between, cast
from sqlalchemy.sql import func, extract

class AdvancedQueries:
    def __init__(self):
        self.Session = sessionmaker(bind=engine)

    def complex_filters(self):
        """复杂过滤条件"""
        with session_scope() as session:
            # 组合条件查询
            results = session.query(User).filter(
                and_(
                    User.is_active == True,
                    or_(
                        User.username.like('%张%'),
                        User.email.contains('example')
                    ),
                    not_(User.id.in_([3, 5, 7]))
                )
            ).all()
            return results

    def date_range_queries(self):
        """日期范围查询"""
        with session_scope() as session:
            # 本周发布的文章
            from datetime import datetime, timedelta
            week_ago = datetime.now() - timedelta(days=7)
            posts = session.query(Post).filter(
                Post.created_at >= week_ago
            ).order_by(desc(Post.created_at)).all()

            # 按月统计
            monthly_stats = session.query(
                extract('year', Post.created_at).label('year'),
                extract('month', Post.created_at).label('month'),
                func.count(Post.id).label('post_count')
            ).group_by('year', 'month').all()
            return monthly_stats

    def pagination(self, page=1, per_page=20):
        """分页查询"""
        with session_scope() as session:
            offset = (page - 1) * per_page
            posts = session.query(Post).order_by(
                desc(Post.created_at)
            ).limit(per_page).offset(offset).all()
            total = session.query(func.count(Post.id)).scalar()
            return {
                'items': posts,
                'total': total,
                'page': page,
                'pages': (total + per_page - 1) // per_page
            }

from sqlalchemy.orm import lazyload, joinedload, subqueryload
from sqlalchemy import update, delete

class BulkOperations:
    def __init__(self):
        self.Session = sessionmaker(bind=engine)

    def bulk_insert(self, data_list):
        """批量插入（高性能）"""
        with session_scope() as session:
            session.bulk_insert_mappings(User, data_list)

    def bulk_update(self):
        """批量更新"""
        with session_scope() as session:
            # 方式 1：使用 update 语句
            session.execute(
                update(User).where(User.is_active == False).values(is_active=True)
            )
            # 方式 2：批量更新映射
            users = session.query(User).filter(User.is_active == True).all()
            for user in users:
                user.last_login = datetime.now()
            session.commit()

    def optimized_query(self):
        """优化关联查询"""
        with session_scope() as session:
            # 使用 joinedload 避免 N+1 查询
            users = session.query(User).options(
                joinedload(User.posts)
            ).filter(User.is_active == True).all()

            # 子查询加载（适用于一对多）
            users = session.query(User).options(
                subqueryload(User.posts)
            ).all()

            # 延迟加载（默认策略）
            posts = session.query(Post).options(
                lazyload(Post.comments)
            ).all()
            return users

from sqlalchemy import event
from datetime import datetime

class EventListeners:
    def __init__(self):
        self.setup_listeners()

    def setup_listeners(self):
        """设置事件监听"""
        @event.listens_for(User, 'before_insert')
        def user_before_insert(mapper, connection, target):
            """用户插入前的钩子"""
            print(f"准备创建用户：{target.username}")
            target.created_at = datetime.utcnow()

        @event.listens_for(User, 'after_insert')
        def user_after_insert(mapper, connection, target):
            """用户插入后的钩子"""
            print(f"用户已创建：{target.username}, ID: {target.id}")
            # 可以在这里发送欢迎邮件、创建默认设置等

        @event.listens_for(Post, 'before_update')
        def post_before_update(mapper, connection, target):
            """文章更新前的钩子"""
            target.updated_at = datetime.utcnow()
            print(f"文章即将更新：{target.id}")

        @event.listens_for(engine, 'before_cursor_execute')
        def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
            """SQL 执行前的钩子（用于审计）"""
            conn.info.setdefault('query_start_time', []).append(datetime.now())
            print(f"执行 SQL: {statement[:100]}...")

        @event.listens_for(engine, 'after_cursor_execute')
        def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
            """SQL 执行后的钩子（性能监控）"""
            total = datetime.now() - conn.info['query_start_time'].pop()
            print(f"SQL 执行耗时：{total.total_seconds():.4f}秒")
            if total.total_seconds() > 0.5:
                print(f"⚠️ 慢查询警告：{total.total_seconds():.2f}秒\n{statement[:200]}")

from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
from sqlalchemy import TypeDecorator
import json

# 自定义 JSON 类型
class JSONEncodedDict(TypeDecorator):
    """支持 JSON 序列化的自定义类型"""
    impl = String

    def process_bind_param(self, value, dialect):
        if value is not None:
            value = json.dumps(value)
        return value

    def process_result_value(self, value, dialect):
        if value is not None:
            value = json.loads(value)
        return value

# 扩展用户模型
class UserExt(Base):
    __tablename__ = 'users_ext'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    first_name = Column(String(50))
    last_name = Column(String(50))
    age = Column(Integer)
    settings = Column(JSONEncodedDict, default={})

    # 存储 JSON 配置
    @hybrid_property
    def full_name(self):
        """计算属性"""
        return f"{self.first_name or ''} {self.last_name or ''}".strip()

    @full_name.expression
    def full_name(cls):
        """在 SQL 查询中使用计算属性"""
        return func.concat(cls.first_name, ' ', cls.last_name)

    @hybrid_method
    def is_adult(self):
        """判断是否成年"""
        return self.age >= 18

    @is_adult.expression
    def is_adult(cls):
        return cls.age >= 18

    @hybrid_property
    def display_name(self):
        """带默认值的显示名称"""
        return self.full_name or self.username

    def get_setting(self, key, default=None):
        """获取用户设置"""
        return self.settings.get(key, default)

from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, Enum, DECIMAL
from sqlalchemy.orm import relationship, validates
from datetime import datetime
import enum

class OrderStatus(enum.Enum):
    PENDING = "待支付"
    PAID = "已支付"
    SHIPPED = "已发货"
    DELIVERED = "已送达"
    CANCELLED = "已取消"
    REFUNDED = "已退款"

class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    sku = Column(String(50), unique=True, nullable=False, index=True)
    name = Column(String(200), nullable=False)
    description = Column(Text)
    price = Column(DECIMAL(10, 2), nullable=False)
    cost = Column(DECIMAL(10, 2))  # 成本价
    stock = Column(Integer, default=0)
    category = Column(String(50), index=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    order_items = relationship("OrderItem", back_populates="product")

    @validates('stock')
    def validate_stock(self, key, value):
        """库存不能为负数"""
        if value < 0:
            raise ValueError("库存不能为负数")
        return value

class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    order_no = Column(String(50), unique=True, nullable=False, index=True)
    user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    status = Column(Enum(OrderStatus), default=OrderStatus.PENDING)
    subtotal = Column(DECIMAL(10, 2), default=0)  # 小计
    discount = Column(DECIMAL(10, 2), default=0)  # 折扣
    shipping_fee = Column(DECIMAL(10, 2), default=0)  # 运费
    total = Column(DECIMAL(10, 2), default=0)  # 总计
    shipping_address = Column(Text, nullable=False)
    recipient_name = Column(String(100))
    recipient_phone = Column(String(20))
    created_at = Column(DateTime, default=datetime.utcnow)
    paid_at = Column(DateTime)
    shipped_at = Column(DateTime)
    delivered_at = Column(DateTime)

    # 关系
    user = relationship("User")
    items = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan")

    @hybrid_property
    def total_items(self):
        """订单商品总数"""
        return sum(item.quantity for item in self.items)

    def calculate_total(self):
        """计算订单总额"""
        self.subtotal = sum(item.subtotal for item in self.items)
        self.total = self.subtotal - self.discount + self.shipping_fee
        return self.total

    def update_status(self, new_status):
        """更新订单状态"""
        self.status = new_status
        if new_status == OrderStatus.PAID:
            self.paid_at = datetime.utcnow()
        elif new_status == OrderStatus.SHIPPED:
            self.shipped_at = datetime.utcnow()
        elif new_status == OrderStatus.DELIVERED:
            self.delivered_at = datetime.utcnow()

class OrderItem(Base):
    __tablename__ = 'order_items'
    id = Column(Integer, primary_key=True)
    order_id = Column(Integer, ForeignKey('orders.id'), nullable=False)
    product_id = Column(Integer, ForeignKey('products.id'), nullable=False)
    product_name = Column(String(200))  # 快照，防止商品信息变更
    unit_price = Column(DECIMAL(10, 2), nullable=False)  # 成交单价
    quantity = Column(Integer, nullable=False)
    subtotal = Column(DECIMAL(10, 2))  # 小计
    order = relationship("Order", back_populates="items")
    product = relationship("Product")

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.subtotal = self.unit_price * self.quantity
        self.product_name = self.product.name

# 订单管理服务类
class OrderService:
    def __init__(self, session):
        self.session = session

    def create_order(self, user_id, items_data, shipping_info):
        """创建订单"""
        from uuid import uuid4
        # 生成订单号
        order_no = f"ORD{datetime.utcnow().strftime('%Y%m%d')}{str(uuid4())[:8]}"
        order = Order(
            order_no=order_no,
            user_id=user_id,
            shipping_address=shipping_info['address'],
            recipient_name=shipping_info['name'],
            recipient_phone=shipping_info['phone']
        )
        self.session.add(order)
        self.session.flush()  # 获取 order.id

        # 添加订单项
        for item_data in items_data:
            product = self.session.query(Product).get(item_data['product_id'])
            if not product or product.stock < item_data['quantity']:
                raise ValueError(f"商品库存不足：{product.name if product else '未知商品'}")
            # 扣减库存
            product.stock -= item_data['quantity']
            order_item = OrderItem(
                order_id=order.id,
                product_id=product.id,
                unit_price=product.price,
                quantity=item_data['quantity'],
                product=product
            )
            self.session.add(order_item)
        order.calculate_total()
        self.session.commit()
        return order

    def get_user_orders(self, user_id, status=None, page=1, per_page=10):
        """获取用户订单列表"""
        query = self.session.query(Order).filter(Order.user_id == user_id)
        if status:
            query = query.filter(Order.status == status)
        # 分页
        offset = (page - 1) * per_page
        orders = query.order_by(desc(Order.created_at)).limit(per_page).offset(offset).all()
        total = query.count()
        return {
            'orders': orders,
            'total': total,
            'page': page,
            'pages': (total + per_page - 1) // per_page
        }

    def cancel_order(self, order_id, user_id=None):
        """取消订单"""
        query = self.session.query(Order).filter(Order.id == order_id)
        if user_id:
            query = query.filter(Order.user_id == user_id)
        order = query.first()
        if not order:
            raise ValueError("订单不存在")
        if order.status not in [OrderStatus.PENDING, OrderStatus.PAID]:
            raise ValueError(f"订单状态为{order.status.value}，无法取消")
        # 恢复库存
        for item in order.items:
            product = self.session.query(Product).get(item.product_id)
            if product:
                product.stock += item.quantity
        order.status = OrderStatus.CANCELLED
        self.session.commit()
        return order

    def get_sales_report(self, start_date, end_date, group_by='day'):
        """销售报表统计"""
        from sqlalchemy import func, case
        query = self.session.query(
            Order.status,
            func.count(Order.id).label('order_count'),
            func.sum(Order.total).label('total_sales'),
            func.avg(Order.total).label('avg_order_value')
        ).filter(
            Order.created_at.between(start_date, end_date)
        ).group_by(Order.status)
        return query.all()

# 使用示例
def ecommerce_example():
    engine = create_engine('sqlite:///./ecommerce.db', echo=False)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()

    # 创建商品
    product1 = Product(
        sku='IPHONE-15-PRO-256',
        name='iPhone 15 Pro 256GB',
        price=8999.00,
        cost=6500.00,
        stock=100,
        category='手机'
    )
    session.add(product1)
    session.commit()

    # 创建订单
    service = OrderService(session)
    order = service.create_order(
        user_id=1,
        items_data=[{'product_id': product1.id, 'quantity': 1}],
        shipping_info={
            'address': '北京市朝阳区 xxx 街道',
            'name': '张三',
            'phone': '13800138000'
        }
    )
    print(f"订单创建成功：{order.order_no}")
    print(f"订单金额：{order.total}")
    session.close()

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean, Table
from sqlalchemy.orm import relationship, validates
from datetime import datetime

# 多对多关系表：文章 - 标签
article_tags = Table(
    'article_tags', Base.metadata,
    Column('article_id', Integer, ForeignKey('articles.id')),
    Column('tag_id', Integer, ForeignKey('tags.id'))
)

class Category(Base):
    __tablename__ = 'categories'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)
    slug = Column(String(50), unique=True, nullable=False)
    description = Column(String(200))
    articles = relationship("Article", back_populates="category")

    def __repr__(self):
        return f"<Category {self.name}>"

class Tag(Base):
    __tablename__ = 'tags'
    id = Column(Integer, primary_key=True)
    name = Column(String(30), unique=True, nullable=False)
    slug = Column(String(30), unique=True, nullable=False)
    articles = relationship("Article", secondary=article_tags, back_populates="tags")

class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    slug = Column(String(200), unique=True, nullable=False, index=True)
    summary = Column(String(500))
    content = Column(Text, nullable=False)
    
    # 状态：草稿、待审核、已发布、已归档
    status = Column(String(20), default='draft', index=True)
    views = Column(Integer, default=0)
    likes = Column(Integer, default=0)
    
    # 外键
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)
    category_id = Column(Integer, ForeignKey('categories.id'))
    
    # 时间戳
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    published_at = Column(DateTime)

    # 关系
    author = relationship("User")
    category = relationship("Category", back_populates="articles")
    tags = relationship("Tag", secondary=article_tags, back_populates="articles")

    @validates('slug')
    def validate_slug(self, key, value):
        """验证 slug 格式"""
        import re
        if not re.match(r'^[a-z0-9-]+$', value):
            raise ValueError("Slug 只能包含小写字母、数字和连字符")
        return value

    def publish(self):
        """发布文章"""
        self.status = 'published'
        self.published_at = datetime.utcnow()

    @hybrid_property
    def reading_time(self):
        """估算阅读时间（分钟）"""
        words_per_minute = 200
        word_count = len(self.content.split())
        return max(1, round(word_count / words_per_minute))

    @classmethod
    def get_popular(cls, session, limit=10):
        """获取热门文章"""
        return session.query(cls).filter(
            cls.status == 'published'
        ).order_by(
            desc(cls.views + cls.likes * 2)  # 综合权重
        ).limit(limit).all()

class ArticleVersion(Base):
    """文章版本历史"""
    __tablename__ = 'article_versions'
    id = Column(Integer, primary_key=True)
    article_id = Column(Integer, ForeignKey('articles.id'), nullable=False)
    version = Column(Integer, nullable=False)
    title = Column(String(200))
    content = Column(Text)
    summary = Column(String(500))
    created_at = Column(DateTime, default=datetime.utcnow)
    created_by = Column(Integer, ForeignKey('users.id'))
    article = relationship("Article")
    user = relationship("User", foreign_keys=[created_by])

# CMS 服务类
class ContentService:
    def __init__(self, session):
        self.session = session

    def create_article(self, author_id, title, content, category_id=None, tags=None):
        """创建文章"""
        from slugify import slugify
        # 生成 slug
        base_slug = slugify(title)
        slug = base_slug
        counter = 1
        while self.session.query(Article).filter(Article.slug == slug).first():
            slug = f"{base_slug}-{counter}"
            counter += 1

        article = Article(
            title=title,
            slug=slug,
            content=content,
            author_id=author_id,
            category_id=category_id,
            status='draft'
        )
        # 提取摘要
        article.summary = content[:200] + "..." if len(content) > 200 else content

        # 添加标签
        if tags:
            for tag_name in tags:
                tag = self.session.query(Tag).filter(Tag.name == tag_name).first()
                if not tag:
                    tag = Tag(
                        name=tag_name,
                        slug=slugify(tag_name)
                    )
                    self.session.add(tag)
                article.tags.append(tag)

        self.session.add(article)
        self.session.commit()
        # 创建版本记录
        self.create_version(article.id, author_id)
        return article

    def update_article(self, article_id, user_id, **kwargs):
        """更新文章"""
        article = self.session.query(Article).get(article_id)
        if not article:
            raise ValueError("文章不存在")
        # 保存旧版本
        self.create_version(article_id, user_id)
        # 更新字段
        for key, value in kwargs.items():
            if hasattr(article, key) and key not in ['id', 'created_at', 'slug']:
                setattr(article, key, value)
        article.updated_at = datetime.utcnow()
        self.session.commit()
        return article

    def create_version(self, article_id, user_id):
        """创建文章版本"""
        article = self.session.query(Article).get(article_id)
        # 获取最新版本号
        latest_version = self.session.query(func.max(ArticleVersion.version)).filter(
            ArticleVersion.article_id == article_id
        ).scalar() or 0
        version = ArticleVersion(
            article_id=article_id,
            version=latest_version + 1,
            title=article.title,
            content=article.content,
            summary=article.summary,
            created_by=user_id
        )
        self.session.add(version)
        self.session.commit()
        return version

    def search_articles(self, query, page=1, per_page=20):
        """全文搜索"""
        from sqlalchemy import or_
        search_query = self.session.query(Article).filter(
            Article.status == 'published',
            or_(
                Article.title.contains(query),
                Article.content.contains(query),
                Article.summary.contains(query)
            )
        ).order_by(desc(Article.published_at))
        total = search_query.count()
        offset = (page - 1) * per_page
        articles = search_query.limit(per_page).offset(offset).all()
        return {
            'articles': articles,
            'total': total,
            'query': query,
            'page': page,
            'pages': (total + per_page - 1) // per_page
        }

    def get_archive(self, year=None, month=None):
        """获取文章归档"""
        query = self.session.query(
            extract('year', Article.published_at).label('year'),
            extract('month', Article.published_at).label('month'),
            func.count(Article.id).label('count')
        ).filter(Article.status == 'published')
        if year:
            query = query.filter(extract('year', Article.published_at) == year)
        if month:
            query = query.filter(extract('month', Article.published_at) == month)
        return query.group_by('year', 'month').order_by('year', 'month').all()

# 使用示例
def cms_example():
    engine = create_engine('sqlite:///./cms.db', echo=False)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    service = ContentService(session)

    # 创建文章
    article = service.create_article(
        author_id=1,
        title='SQLAlchemy 高级特性详解',
        content='本文深入探讨 SQLAlchemy 的事件系统、混合属性等高级特性...',
        tags=['SQLAlchemy', 'Python', '数据库']
    )
    # 发布文章
    article.publish()
    session.commit()

    # 搜索文章
    results = service.search_articles('SQLAlchemy')
    print(f"找到 {results['total']} 篇相关文章")
    session.close()

SQLAlchemy 核心用法与实战案例解析

一、库的简介

二、安装 SQLAlchemy

三、基本用法：四步掌握 SQLAlchemy

1. 创建引擎和连接

2. 定义数据模型

3. 会话管理和 CRUD 操作

4. 上下文管理器与资源管理

四、高级用法

1. 复杂的查询表达式

2. 批量操作与性能优化

3. 事件监听与钩子函数

4. 混合属性与自定义数据类型

五、实际应用场景

场景一：电商平台订单系统

场景二：内容管理系统（CMS）

更多推荐文章

相关免费在线工具

SQLAlchemy 核心用法与实战案例解析

一、库的简介

二、安装 SQLAlchemy

三、基本用法：四步掌握 SQLAlchemy

1. 创建引擎和连接

2. 定义数据模型

3. 会话管理和 CRUD 操作

4. 上下文管理器与资源管理

四、高级用法

1. 复杂的查询表达式

2. 批量操作与性能优化

3. 事件监听与钩子函数

4. 混合属性与自定义数据类型

五、实际应用场景

场景一：电商平台订单系统

场景二：内容管理系统（CMS）

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具