Python面试题：结合Python技术，如何使用SQLAlchemy进行数据库操作

使用SQLAlchemy进行数据库操作是一种非常灵活和强大的方式，它支持多种关系型数据库，并提供了ORM（对象关系映射）和核心（SQL Expression Language）两种使用方法。以下是详细的步骤和示例，展示如何在Python中使用SQLAlchemy进行数据库操作。

1. 安装SQLAlchemy

首先，确保你已经安装了SQLAlchemy和相应的数据库驱动程序。例如，对于SQLite，你只需要安装SQLAlchemy，但对于PostgreSQL，你需要安装psycopg2。

pip install sqlalchemy
pip install psycopg2-binary  # 如果使用PostgreSQL

2. 创建数据库连接

使用SQLAlchemy创建数据库连接和会话。

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmakerDATABASE_URL = "postgresql://user:password@localhost/dbname"  # 替换为你的数据库URLengine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

3. 定义模型

使用SQLAlchemy的ORM定义模型。

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()class Item(Base):__tablename__ = 'items'id = Column(Integer, primary_key=True, index=True)name = Column(String, index=True)description = Column(String, index=True)price = Column(Float)tax = Column(Float)

4. 创建数据库表

创建表结构：

Base.metadata.create_all(bind=engine)

5. 数据库操作

使用会话进行增删改查操作。

创建会话

db = SessionLocal()

插入数据

new_item = Item(name="Test Item", description="A test item", price=10.99, tax=1.50)
db.add(new_item)
db.commit()
db.refresh(new_item)  # 获取插入后的数据，包括自增ID
print(new_item.id)

查询数据

items = db.query(Item).all()
for item in items:print(item.name, item.price)# 查询单个记录
item = db.query(Item).filter(Item.id == 1).first()
print(item.name, item.price)

更新数据

item = db.query(Item).filter(Item.id == 1).first()
item.price = 12.99
db.commit()

删除数据

item = db.query(Item).filter(Item.id == 1).first()
db.delete(item)
db.commit()

6. 异步支持（使用SQLAlchemy 1.4+）

SQLAlchemy 1.4+ 支持异步操作。需要使用 asyncio 和 asyncpg 驱动。

pip install asyncpg
pip install sqlalchemy[asyncio]

创建异步引擎和会话：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmakerDATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)

异步操作示例：

import asyncioasync def main():async with AsyncSessionLocal() as session:async with session.begin():new_item = Item(name="Async Item", description="An async item", price=10.99, tax=1.50)session.add(new_item)await session.commit()async with session.begin():result = await session.execute(select(Item).where(Item.name == "Async Item"))item = result.scalars().first()print(item.name, item.price)asyncio.run(main())

7. 使用SQL Expression Language

除了ORM，SQLAlchemy还提供了SQL Expression Language用于构建原生SQL查询。

from sqlalchemy import Table, MetaData, selectmetadata = MetaData()
items_table = Table('items', metadata, autoload_with=engine)with engine.connect() as conn:query = select([items_table]).where(items_table.c.name == 'Test Item')result = conn.execute(query)for row in result:print(row)