FastApi操作MySQL基本文件结构

首先是总的文件结构：

.

├── __pycache__

│   └── main.cpython-38.pyc

├── main.py

└── sqlstu

    ├── __pycache__

    │   ├── crud.cpython-38.pyc

    │   ├── database.cpython-38.pyc

    │   └── models.cpython-38.pyc

    ├── crud.py

    ├── database.py

    └── models.py

main.py就是FastAPI的启动文件了。

main.py:

import uvicorn
from pydantic import BaseModel
from sqlstu import crud
from sqlalchemy.orm import Session
from fastapi import Response, Depends, FastAPI, HTTPException
from sqlstu.database import SessionLocal, engine
from sqlstu import models

app = FastAPI()

models.Base.metadata.create_all(bind=engine)


# 数据库开关
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


class Item(BaseModel):
    form: str
    form_name: str
    entry: dict


@app.get('/')
async def index(db: Session = Depends(get_db)):
    a = '111'
    b = '222'
    crud.add_record(db, a, b)
    return 1


if __name__ == '__main__':
    uvicorn.run(app='main:app', host='0.0.0.0', port=8000, reload=True, debug=False)

crud.py进行数据库增删改查

from sqlalchemy.orm import Session
from .models import *


def add_record(db: Session, source, result):
    if not db.query(PandaParse).filter(PandaParse.source == source).first():
        added = PandaParse(source=source, result=result)
        db.add(added)
        db.commit()
        db.refresh(added)
    return 200

database.py建立数据库链接

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://username:password@localhost:port/database"

engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

models.py数据库表对应文件

from sqlalchemy import Boolean, Column, Integer, String, Datetime
from datetime import datetime
from .database import Base


class PandaParse(Base):
    __tablename__ = "LanAPI_Panda"
    id = Column(Integer, primary_key=True, index=True)
    source = Column(String)
    result = Column(String)