95 lines
3.0 KiB
Python
95 lines
3.0 KiB
Python
from fastapi import FastAPI, HTTPException, Depends
|
||
from fastapi.responses import JSONResponse
|
||
from sqlalchemy.orm import Session
|
||
from typing import List, Optional,Any
|
||
|
||
from app import models, schemas, crud
|
||
from app.database import SessionLocal, engine
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
|
||
# 初始化数据库
|
||
models.Base.metadata.create_all(bind=engine)
|
||
|
||
app = FastAPI()
|
||
|
||
# 允许所有来源跨域访问（不推荐用于生产环境）
|
||
origins = ["*"]
|
||
|
||
# 设置CORS（跨域资源共享）策略
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"], # 允许所有来源
|
||
allow_credentials=True,
|
||
allow_methods=["*"], # 允许所有方法
|
||
allow_headers=["*"], # 允许所有请求头
|
||
)
|
||
# 依赖注入
|
||
def get_db():
|
||
db = SessionLocal()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
|
||
|
||
@app.get("/")
|
||
def root():
|
||
return {"message": "Hello World"}
|
||
|
||
@app.get("/hello")
|
||
def hello():
|
||
return {"message": "Hello World"}
|
||
|
||
# @app.get("/add", response_model=schemas.User)
|
||
@app.get("/add")
|
||
|
||
def add_user(
|
||
nickname: Optional[Any] = None,
|
||
remark: Optional[Any] = None,
|
||
wechat_number: Optional[Any] = None,
|
||
wxid: Optional[Any] = None,
|
||
gender: Optional[Any] = None,
|
||
friendaddtime: Optional[Any] = None,
|
||
openid: Optional[Any] = None,
|
||
db: Session = Depends(get_db)
|
||
):
|
||
crud.create_user(db=db, user=schemas.UserCreate(
|
||
nickname=nickname, remark=remark, wechat_number=wechat_number,
|
||
wxid=wxid, gender=gender, friendaddtime=friendaddtime,
|
||
openid=openid))
|
||
response=f'''{nickname}你好,\n我是机器人助理\n一名开发者,副业:跨境电商,#公众号:异度世界'''
|
||
# content = {
|
||
# "answer": response,
|
||
# "answer_type": "text",
|
||
# }
|
||
headers = {"Content-Type": "application/json"}
|
||
return JSONResponse(content=response, headers=headers)
|
||
# return f'''{nickname}你好,\n我是机器人助理\n一名开发者,副业:跨境电商,#公众号:异度世界'''
|
||
|
||
@app.get("/users/")
|
||
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
|
||
return crud.get_users(db=db, skip=skip, limit=limit)
|
||
|
||
@app.get("/user/{identifier}")
|
||
def read_user(identifier: str, db: Session = Depends(get_db)):
|
||
user = crud.get_user_by_openid_or_wxid(db, identifier=identifier)
|
||
if user is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return user
|
||
|
||
@app.delete("/user/{user_id}")
|
||
def delete_user(user_id: int, db: Session = Depends(get_db)):
|
||
user = crud.get_user(db=db, user_id=user_id)
|
||
if user is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return crud.delete_user(db=db, user=user)
|
||
|
||
@app.put("/user/{identifier}")
|
||
def update_user(identifier: str, user_update: schemas.UserUpdate, db: Session = Depends(get_db)):
|
||
user = crud.get_user_by_openid_or_wxid(db, identifier=identifier)
|
||
if user is None:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
|
||
return crud.update_user(db=db, user=user, user_update=user_update) |