python的web接口数据库链接封装
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pymysql import connect, cursors
from pymysql.cursors import DictCursor
app = FastAPI()# 设置允许的源列表
origins = ["http://localhost","http://localhost:8081","http://localhost:3000","https://your-production-domain.com"
]# 添加中间件
app.add_middleware(CORSMiddleware,allow_origins=origins, # 也可以设置为 ["*"] 允许所有源allow_credentials=True,allow_methods=["*"], # 允许所有方法allow_headers=["*"], # 允许所有头
)
class Database:def __init__(self):self.config = {'host': 'localhost','user': 'root','password': 'admin','database': 'demo','cursorclass': cursors.DictCursor}self.connection = Nonedef __enter__(self):"""进入上下文时创建连接"""self.connection = connect(**self.config)return self.connection.cursor()def __exit__(self, exc_type, exc_val, exc_tb):"""退出上下文时处理连接"""if self.connection:if exc_type is None:self.connection.commit()else:self.connection.rollback()self.connection.close()self.connection = None@app.get('/')
def main():return {"hello": "wo1rld"}@app.get('/demo')
def demo():with Database() as cursor:cursor.execute("SELECT * from user limit 2")data = cursor.fetchall()return list(data)@app.get('/test/a={a}/b={b}')
def test(a: str=None, b: str=None):with Database() as cursor:cursor.execute(f"SELECT * from user where id={a}")data1 = cursor.fetchall()return list(data1)if __name__ == '__main__':import uvicornuvicorn.run(app=app,host="0.0.0.0",port=8080,workers=1)访问地址：127.0.0.1:8080/test/a=1/b=4
或者127.0.0.1:8080/demo