SQLAlchemy 的Core与ORM

✅ 环境准备

假设你已经定义了如下模型和数据库连接：

1. 配置数据库连接

# db.py

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

DATABASE_URL = "sqlite:///./test.db"  # 示例用 SQLite

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

2. 定义 ORM 模型

# models.py

from sqlalchemy import Column, Integer, String
from db import Base

class User(Base):
__tablename__ = "users"

id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)

✅ 创建表

# main.py
from db import engine
from models import Base

Base.metadata.create_all(bind=engine)

✅ 增删改查（CRUD）

导入准备

from db import SessionLocal
from models import User

✅ 1. 添加用户（Create）

def create_user(name: str, email: str):
db = SessionLocal()
try:
new_user = User(name=name, email=email)
db.add(new_user)
db.commit()
db.refresh(new_user)  # 刷新以获取 ID 等信息
return new_user
finally:
db.close()

✅ 2. 查询用户（Read）

def get_user_by_id(user_id: int):
db = SessionLocal()
try:
return db.query(User).filter(User.id == user_id).first()
finally:
db.close()

def get_all_users():
db = SessionLocal()
try:
return db.query(User).all()
finally:
db.close()

✅ 3. 更新用户（Update）

def update_user(user_id: int, new_name: str):
db = SessionLocal()
try:
user = db.query(User).filter(User.id == user_id).first()
if user:
user.name = new_name
db.commit()
db.refresh(user)
return user
finally:
db.close()

✅ 4. 删除用户（Delete）

def delete_user(user_id: int):
db = SessionLocal()
try:
user = db.query(User).filter(User.id == user_id).first()
if user:
db.delete(user)
db.commit()
return user
finally:
db.close()

🔁 总结表

| 操作 | 方法 |
| -- | ---------------------------------------- |
| 增 | db.add() → db.commit() |
| 查 | db.query().filter().first() 或 .all() |
| 改 | 修改字段 → db.commit() → db.refresh() |
| 删 | db.delete() → db.commit() |

如果你在使用 FastAPI，我可以进一步演示如何结合依赖注入把这些 CRUD 操作集成到路由中。需要的话可以继续问。

在 SQLAlchemy ORM 中，多条件查询可以通过 .filter() 方法，结合 and_()、or_() 或多个 .filter() 链式调用来实现。

下面是一些常用的多条件查询方式示例。

✅ 示例模型（假设如下）

class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String)
email = Column(String)
age = Column(Integer)

✅ 方式一：多个 .filter() 连用（推荐）

user = db.query(User).filter(User.name == "Alice").filter(User.age > 20).all()

等价于 SQL：

SELECT * FROM users WHERE name = 'Alice' AND age > 20;

✅ 方式二：使用 and_ 明确表达多个条件

from sqlalchemy import and_

user = db.query(User).filter(and_(User.name == "Alice", User.age > 20)).all()

✅ 方式三：使用 or_ 进行“或”查询

from sqlalchemy import or_

user = db.query(User).filter(or_(User.name == "Alice", User.name == "Bob")).all()

等价于：

SELECT * FROM users WHERE name = 'Alice' OR name = 'Bob';

✅ 方式四：混合使用 and_ 与 or_

from sqlalchemy import and_, or_

user = db.query(User).filter(
and_(
User.age > 18,
or_(User.name == "Alice", User.name == "Bob")
)
).all()

SQL：

SELECT * FROM users WHERE age > 18 AND (name = 'Alice' OR name = 'Bob');

✅ 可选：通过变量传参构造查询

conditions = []
if name:
conditions.append(User.name == name)
if min_age:
conditions.append(User.age >= min_age)

user = db.query(User).filter(*conditions).all()

🔄 总结

在 SQLAlchemy Core 模式下，你可以使用 text() 或者 select() / insert() / update() / delete() 等构建块，来执行 自定义 SQL 语句。

✅ 一、使用 text() 执行自定义 SQL（完全手写 SQL）

from sqlalchemy import text
from config.db import engine

with engine.connect() as conn:
result = conn.execute(text("SELECT * FROM users WHERE age > :age"), {"age": 18})
rows = result.fetchall()
for row in rows:
print(row)

✅ 特点：

• 灵活，可写任意 SQL（包括复杂 JOIN、子查询等）
• 使用参数绑定（防止 SQL 注入）

✅ 二、使用 select() + Core 语法构建 SQL

from sqlalchemy import select
from models.core_model import users  # 假设 core 中定义了 Table: users
from config.db import engine

with engine.connect() as conn:
stmt = select(users).where(users.c.age > 18)
result = conn.execute(stmt)
for row in result.fetchall():
print(row)

✅ 三、INSERT 自定义数据

from sqlalchemy import insert

stmt = insert(users).values(name="Alice", age=22)
with engine.connect() as conn:
conn.execute(stmt)
conn.commit()

✅ 四、UPDATE 语句

from sqlalchemy import update

stmt = update(users).where(users.c.id == 1).values(name="UpdatedName")
with engine.connect() as conn:
conn.execute(stmt)
conn.commit()

✅ 五、DELETE 语句

from sqlalchemy import delete

stmt = delete(users).where(users.c.id == 1)
with engine.connect() as conn:
conn.execute(stmt)
conn.commit()

✅ 六、打印构建的 SQL（调试用）

print(stmt.compile(compile_kwargs={"literal_binds": True}))

🔁 总结