.
└── sql_app
├── __init__.py
├── curd.py
├── database.py
├── main.py
├── models.py
└── schemas.py
pip install sqlalchemy
# 1、导入 sqlalchemy 部分的包
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 2、声明 database url
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
# 3、创建 sqlalchemy 引擎
engine = create_engine(
url=SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
# 4、创建一个 database 会话
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 5、返回一个 ORM Model
Base = declarative_base()
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
第一行是 slite 连接 url
# sqlite-pysqlite 库
sqlite+pysqlite:///file_path
# mysql-mysqldb 库
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
# mysql-pymysql 库
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
# mysql-mysqlconnector 库
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
# oracle-cx_Oracle 库
oracle+cx_oracle://user:pass@hostname:port[/dbname][?service_name=<service>[&key=value&key=value...]]
# postgresql-pypostgresql 库
postgresql+pypostgresql://user:password@host:port/dbname[?key=value&key=value...]
# SQL Server-PyODBC 库
mssql+pyodbc://<username>:<password>@<dsnname>
engine = create_engine(
url=SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
{"check_same_thread": False}
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
后面会通过继承这个 Base 类,来创建每个数据库 Model,也称为 ORM Model
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base
class User(Base):
# 1、表名
__tablename__ = "users"
# 2、类属性,每一个都代表数据表中的一列
# Column 就是列的意思
# Integer、String、Boolean 就是数据表中,列的类型
id = Column(Integer, primary_key=True, index=True, default=1, autoincrement=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
items = relationship("Item", back_populates="owner")
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True, default=1, autoincrement=True)
title = Column(String, index=True)
description = Column(String, index=True)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="items")
列,一个属性代表数据表中的一列
参数 | 作用 |
---|---|
primary_key | 如果设为 True ,这列就是表的主键 |
unique | 如果设为 True ,这列不允许出现重复的值 |
index | 如果设为 True ,为这列创建索引,提升查询效率 |
nullable | 如果设为 True ,这列允许使用空值; 如果设为 False ,这列不允许使用空值 |
default | 为这列定义默认值 |
autoincrement | 如果设为 True ,这列自增 |
default 为这列定义默认值 autoincrement 如果设为 True ,这列自增
代表数据表中每一列的数据类型
为了避免混淆 SQLAlchemy 模型和 Pydantic 模型之间,将使用文件 models.py 编写 SQLAlchemy 模型和文件 schemas.py 编写 Pydantic 模型
from typing import List, Optional
from pydantic import BaseModel
# Item 的基类,表示创建和查询 Item 时共有的属性
class ItemBase(BaseModel):
title: str
description: Optional[str] = None
# 创建 Item 时的 Model
class ItemCreate(ItemBase):
pass
# 查询 Item 时的 Model
class Item(ItemBase):
id: int
owner_id: int
# 向 Pydantic 提供配置
class Config:
# orm_mode 会告诉 Pydantic 模型读取数据,即使它不是字典,而是 ORM 模型(或任何其他具有属性的任意对象)
orm_mode = True
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
is_active: bool
items: List[Item] = []
class Config:
orm_mode = True
基类,声明在创建或读取数据时共有的属性
创建数据时使用的 Model
读取数据时使用的 Model
class Config:
orm_mode = True
# 正常情况
id = data["id"]
# 还会尝试从对象获取属性
id = data.id
设置了 orm_mode,Pydantic 模型与 ORM 就兼容了,只需在路径操作的 response_model 参数中声明它即可
current_user.items
代码只实现了查询和创建
from sqlalchemy.orm import Session
from .models import User, Item
from .schemas import UserCreate, ItemCreate
# 根据 id 获取 user
def get_user(db: Session, user_id: int):
return db.query(User).filter(User.id == user_id).first()
# 根据 email 获取 user
def get_user_by_email(db: Session, email: str):
return db.query(User).filter(User.email == email).first()
# 获取所有 user
def get_users(db: Session, size: int = 0, limit: int = 100):
return db.query(User).offset(size).limit(limit).all()
# 创建 user,user 类型是 Pydantic Model
def create_user(db: Session, user: UserCreate):
fake_hashed_password = user.password + "superpolo"
# 1、使用传进来的数据创建 SQLAlchemy Model 实例对象
db_user = User(email=user.email, hashed_password=fake_hashed_password)
# 2、将实例对象添加到数据库会话 Session 中
db.add(db_user)
# 3、将更改提交到数据库
db.commit()
# 4、刷新实例,方便它包含来自数据库的任何新数据,比如生成的 ID
db.refresh(db_user)
return db_user
# 获取所有 item
def get_items(db: Session, size: int = 0, limit: int = 100):
return db.query(Item).offset(size).limit(limit).all()
# 创建 item,item 类型是 Pydantic Model
def create_item(db: Session, item: ItemCreate, user_id: int):
db_item = Item(**item.dict(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
函数内的操作步骤如下
# 1、使用传进来的数据创建 SQLAlchemy Model 实例对象
db_user = User(email=user.email, hashed_password=fake_hashed_password)
# 2、将实例对象添加到数据库会话 Session 中
db.add(db_user)
# 3、将更改提交到数据库
db.commit()
# 4、刷新实例,方便它包含来自数据库的任何新数据,比如生成的 ID
db.refresh(db_user)
from typing import List
import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status, Path, Query, Body
from sqlalchemy.orm import Session
from models import Base
from schemas import User, UserCreate, ItemCreate, Item
from database import SessionLocal, engine
import curd
Base.metadata.create_all(bind=engine)
app = FastAPI()
# 依赖项,获取数据库会话对象
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# 创建用户
@app.post("/users", response_model=User)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
# 1、先查询用户是否有存在
db_user = curd.get_user_by_email(db, user.email)
if db_user:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="user 已存在")
res_user = curd.create_user(db, user)
return res_user
# 根据 user_id 获取用户
@app.get("/user_id/{user_id}", response_model=User)
async def get_user(user_id: int = Path(...), db: Session = Depends(get_db)):
return curd.get_user(db, user_id)
# 根据 email 获取用户
@app.get("/user_email/{email}", response_model=User)
async def get_user_by_email(email: str = Path(...), db: Session = Depends(get_db)):
return curd.get_user_by_email(db, email)
# 获取所有用户
@app.get("/users_all/", response_model=List[User])
async def get_users(skip: int = Query(0),
limit: int = Query(100),
db: Session = Depends(get_db)):
return curd.get_users(db, skip, limit)
# 创建 item
@app.post("/users/{user_id}/items", response_model=Item)
async def get_user_item(user_id: int = Path(...), item: ItemCreate = Body(...), db: Session = Depends(get_db)):
return curd.create_user_item(db, item, user_id)
# 获取所有 item
@app.get("/items/", response_model=List[Item])
async def get_items(skip: int = Query(0),
limit: int = Query(100),
db: Session = Depends(get_db)):
return curd.get_items(db, skip, limit)
if __name__ == "__main__":
uvicorn.run(app="main:app", host="127.0.0.1", port=8080, reload=True, debug=True)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
async def create_user(user: UserCreate, db: Session = Depends(get_db))
db: Session
# 中间件
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
# 默认响应
response = Response("Internal server error", status_code=500)
try:
request.state.db = SessionLocal()
response = await call_next(request)
finally:
# 关闭数据库会话
request.state.db.close()
return response
# 依赖项,获取数据库会话对象
def get_db(request: Request):
return request.state.db