首先是总的文件结构:
.
├── __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://root:20010714Liu!@cdb-2u2j2ex0.cd.tencentcdb.com:10034/hngcZhaoPin" 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)
发表评论