102 lines
2.7 KiB
Python
102 lines
2.7 KiB
Python
from fastapi import FastAPI,Header,Depends,HTTPException
|
|
import requests
|
|
import json
|
|
from enum import Enum
|
|
from typing import Union
|
|
from pydantic import BaseModel
|
|
|
|
|
|
from sqlalchemy.orm import Session
|
|
import crud, models, schemas
|
|
from database import SessionLocal, engine
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
app = FastAPI()
|
|
|
|
# Dependency
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
class User(BaseModel):
|
|
user_id: str
|
|
|
|
class ModelName(str, Enum):
|
|
alexnet = "alexnet"
|
|
resnet = "resnet"
|
|
lenet = "lenet"
|
|
|
|
class Item(BaseModel):
|
|
name: str
|
|
description: Union[str, None] = None
|
|
price: float
|
|
tax: Union[float, None] = None
|
|
|
|
@app.post("/delete_user")
|
|
async def delete(user:User,db: Session = Depends(get_db)):
|
|
db_user = crud.delete_user(db,user_id=user.user_id)
|
|
# print('db_user'+db_user)
|
|
return {"message": "Hello delete_user"}
|
|
|
|
|
|
@app.post("/update_user")
|
|
async def update(user:User,db: Session = Depends(get_db)):
|
|
db_user = crud.update_user(db,user_id=user.user_id)
|
|
return {"message": "Hellupdate_usero World"}
|
|
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
@app.post("/users/", response_model=schemas.User)
|
|
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
|
|
db_user = crud.get_user_by_email(db, email=user.email)
|
|
if db_user:
|
|
raise HTTPException(status_code=400, detail="Email already registered")
|
|
return crud.create_user(db=db, user=user)
|
|
|
|
# 飞书通知
|
|
@app.get("/feishu")
|
|
async def root():
|
|
headers = {'Content-Type': 'application/json'}
|
|
# https://open.feishu.cn/open-apis/bot/v2/hook/5fb1b44c-3446-42ef-9b3a-dd37731567dd
|
|
try:
|
|
r = requests.post('https://open.feishu.cn/open-apis/bot/v2/hook/5fb1b44c-3446-42ef-9b3a-dd37731567dd',
|
|
data=json.dumps({'msg_type': 'text', "content": {"text": "cityNew有新用户注册"}}), headers=headers)
|
|
except(e):
|
|
print('e')
|
|
return {"message": "触发飞书webhook"}
|
|
|
|
# post请求
|
|
@app.post("/testPost")
|
|
async def root():
|
|
return {"message": "test-post"}
|
|
|
|
# 路径参数
|
|
@app.get("/path/{pathName}")
|
|
async def root(pathName:ModelName):
|
|
return {"message": pathName}
|
|
|
|
# 查询参数
|
|
@app.get("/items/")
|
|
async def read_item(skip: int = 0, limit: Union[int, None] = None):
|
|
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
|
|
# return fake_items_db[skip : skip + limit]
|
|
return f'{skip}+{limit}'
|
|
|
|
# 请求体
|
|
@app.post("/postitems/")
|
|
async def create_item(item: Item):
|
|
return item
|
|
|
|
|
|
# header参数
|
|
@app.get("/items/header")
|
|
async def read_items(user_agent: Union[str, None] = Header(default=None)):
|
|
return {"User-Agent": user_agent} |