FastAPI+SQLAlchemy数据库连接
目录
FastAPI+SQLAlchemy数据库连接 配置数据库连接 创建表模型 创建alembic迁移文件 安装+初始化 编辑env.py 编辑alembic.ini 迁移数据库
视图函数查询
配置数据库连接
from sqlalchemy import create_engine
from sqlalchemy. orm import sessionmaker
DATABASE_URL = "mysql+pymysql://root:7997@localhost/fastapidemo"
engine = create_engine(
DATABASE_URL,
pool_size= 10 ,
max_overflow= 20 ,
pool_timeout= 30 ,
pool_recycle= 1800 ,
)
SessionLocal = sessionmaker( autocommit= False , autoflush= False , bind= engine)
创建表模型
from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy. orm import DeclarativeBase, relationship
class Base ( DeclarativeBase) :
__abstract__ = True
@classmethod
def get_all ( cls, session) :
return session. query( cls) . all ( )
book_press = Table( 'book_press' , Base. metadata,
Column( 'book_id' , Integer, ForeignKey( 'book.id' ) , primary_key= True ) ,
Column( 'press_id' , Integer, ForeignKey( 'press.id' ) , primary_key= True ) )
class Book ( Base) :
__tablename__ = 'book'
id = Column( Integer, primary_key= True , autoincrement= True )
title = Column( String( 15 ) , nullable= False , unique= True )
author_id = Column( Integer, ForeignKey( 'author.id' ) )
author = relationship( 'Author' , backref= 'books' , cascade= 'all,delete' )
press = relationship( 'Press' , backref= 'books' , secondary= book_press)
class Author ( Base) :
__tablename__ = 'author'
id = Column( Integer, primary_key= True , autoincrement= True )
name = Column( String( 15 ) , nullable= False , unique= True )
age = Column( Integer, nullable= False )
class Press ( Base) :
__tablename__ = 'press'
id = Column( Integer, primary_key= True , autoincrement= True )
name = Column( String( 15 ) , nullable= False , unique= True )
if __name__ == '__main__' :
from db import engine
Base. metadata. create_all( engine)
创建alembic迁移文件
安装+初始化
pip install alembic
alembic init alembic
编辑env.py
from logging. config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
from module import Base, Book, Author, Press
config = context. config
target_metadata = Base. metadata
编辑alembic.ini
# alembic.ini
sqlalchemy.url = mysql+pymysql://root:7997@localhost/fastapidemo
迁移数据库
alembic revision --autogenerate
alembic upgrade head
视图函数查询
from fastapi import FastAPI, Request, Depends
from pydantic import BaseModel
from sqlalchemy. orm import Session
from db import SessionLocal
from orm import Author
app = FastAPI( )
session = SessionLocal( )
def get_db ( ) :
db = SessionLocal( )
try :
yield db
finally :
db. close( )
class Item ( BaseModel) :
pass
@app. get ( "/api/" )
async def root ( db: Session = Depends( get_db) ) :
res = db. query( Author) . all ( )
res2 = Author. get_all( db)
print ( res[ 0 ] . name)
print ( res2[ 0 ] . name)
return { f"作者名: { res[ 0 ] . name} , { res2[ 0 ] . name} " }