所以我试图在这里编写简单的函数,但是每次我运行 swagger 时,都会出现上述错误。
这是我的功能:
def authenticate_user(username: str, password: str, db: Session = Depends(bd.get_db)):
user = db.query(bd.User.username).filter(username == username).first()
if not user:
return False
if not verify_password(password, user.password_hash):
return False
return user
Run Code Online (Sandbox Code Playgroud)
这是我的 get_db 函数,它非常标准:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Run Code Online (Sandbox Code Playgroud)
我注意到它Depends(bd.get_db)在端点函数(带有@app.post/@app.get装饰器的函数)中工作得很好,但在普通函数中却不起作用。
显然,我不太理解依赖注入的概念,但我还不能完全理解它。
我有 FastAPI 应用程序在 docker docker 容器中运行。它运行良好,除了一件事之外,如果源代码中进行任何更改,应用程序不会重新加载。仅当容器重新启动时,更改才会应用。
这是我的应用程序的源代码,那么如何确保每次对源代码进行一些更改时我的容器化应用程序都会自动重新加载?
主要.py
from typing import Optional
import uvicorn
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
return {"item_id": item_id, "q": q}
if __name__ == '__main__':
uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
Run Code Online (Sandbox Code Playgroud)
docker-compose.yml
version: "3"
services:
web:
build: .
restart: always
command: bash -c "uvicorn main:app --host 0.0.0.0 --port 8000 --reload"
volumes:
- .:/code
ports:
- "8000:8000"
depends_on:
- db
db:
image: postgres
ports: …Run Code Online (Sandbox Code Playgroud) 在同步而非非模式下使用 FastAPI async,我希望能够接收 POST 请求的原始、未更改的正文。
我能找到的所有示例都显示async代码,当我以正常同步方式尝试时,它们request.body()显示为协程对象。
XML当我通过向此端点发布一些内容来测试它时,我得到了一个500 "Internal Server Error".
from fastapi import FastAPI, Response, Request, Body
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.post("/input")
def input_request(request: Request):
# how can I access the RAW request body here?
body = request.body()
# do stuff with the body here
return Response(content=body, media_type="application/xml")
Run Code Online (Sandbox Code Playgroud)
这对于 FastAPI 来说是不可能的吗?
注意:简化的输入请求如下所示:
POST http://127.0.0.1:1083/input
Content-Type: application/xml
<XML>
<BODY>TEST</BODY>
</XML>
Run Code Online (Sandbox Code Playgroud)
我无法控制输入请求的发送方式,因为我需要替换现有的 SOAP API。
抱歉,不精通Python。
我还没有找到该用例的文档。我如何获取请求正文,确保它是一个有效的 Json(任何有效的 json,包括数字、字符串、布尔值和空值,而不仅仅是对象和数组)并获取实际的 Json。使用 pydantic 强制 Json 具有特定的结构。
def get_db():
db = SessionLocal()
try:
return db
finally:
db.close()
Run Code Online (Sandbox Code Playgroud)
我剪掉了这段代码,以便使用 Sqlalchemy 在 fastapi 中获取 Sessionlocal。好吧,当我使用 return 而不是 Yield 时。我的代码仍然有效。然后,我不明白使用 Yield 的原因。有人能帮我吗?
我目前正在复杂系统上使用 FastAPI 进行 POC。该项目业务逻辑繁重,完成后将与 50 多个不同的数据库表进行交互。每个模型都有一个服务,一些更复杂的业务逻辑有自己的服务(然后通过特定于模型的服务与不同的表进行交互/查询)。
虽然一切正常,但我的团队的一些成员对 Session 对象的依赖注入提出了一些反对。最大的问题主要是必须将会话从控制器传递到服务、第二个服务以及(在少数情况下)进一步的第三个服务。在这些情况下,中间服务功能往往没有数据库查询但他们调用其他服务的功能可能有一些。抱怨主要在于这更难以维护,并且必须到处传递数据库对象似乎毫无用处的重复。
代码示例:
Databases/mysql.py(项目中的 3 个数据库之一)
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
def get_uri():
return 'the mysql uri'
engine = create_engine(get_uri())
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db: Session = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
finally:
db.close()
Run Code Online (Sandbox Code Playgroud)
控制器/controller1.py
from fastapi import APIRouter, HTTPException, Path, Depends
from sqlalchemy.orm import Session
from services.mysql.bar import get_bar_by_id
from services.mysql.process_x import bar_process
from databases.mysql …Run Code Online (Sandbox Code Playgroud) 我正在为我的 ML 模型使用 FAST API。
我有一个管道。
lr_tfidf = Pipeline([('vect', tfidf),
('clf', LogisticRegression(penalty='l2'))])
Run Code Online (Sandbox Code Playgroud)
现在在 Fast API 中,当我想要预测并将结果显示为 API 时,我的代码是
app = FastAPI()
@app.post('/predict')
def predict_species(data: str):
data = np.array([data])
prob = lr_tfidf.predict_proba(data).max()
pred = lr_tfidf.predict(data)
return {'Probability': f'{prob}',
'Predictions':f'{pred}'}
Run Code Online (Sandbox Code Playgroud)
我是从教程里复制过来的。当我通过 FASTAPI 在 GUI 上测试它时,它工作得很好,如图所示,即它显示了概率和预测。
当我转到 GUI 提供的请求 URL 时http://127.0.0.1:8000/predict?data=hello(测试数据是 hello),它给了我错误。
{"detail":"Method Not Allowed"}
Run Code Online (Sandbox Code Playgroud)
在我的终端上,错误消息是
INFO: 127.0.0.1:42568 - "GET /predict?data=hello HTTP/1.1" 405 Method Not Allowed
Run Code Online (Sandbox Code Playgroud) 无论如何,FastAPI“依赖项”是否可以解释路径参数?
我有很多形式的函数:
@app.post("/item/{item_id}/process", response_class=ProcessResponse)
async def process_item(item_id: UUID, session: UserSession = Depends(security.user_session)) -> ProcessResponse:
item = await get_item(client_id=session.client_id, item_id=item_id)
await item.process()
Run Code Online (Sandbox Code Playgroud)
一次又一次,我需要传入[多个]参数来获取所需的项目,然后再对其进行操作。这是非常重复的并且使得代码非常冗长。我真正想做的是将itemin 作为参数传递给该方法。
理想情况下,我想建立get_item一个依赖项或以某种方式将其嵌入到路由器中。这将大大减少重复的逻辑和过于冗长的函数参数。问题在于客户端在路径中传递了一些关键参数。
是否可以将 Path 参数传递到依赖项中,或者可能在路由器中执行依赖项并传递结果?
我正在尝试http://httpbin.org/uuid使用以下代码片段一次向服务器发送 100 个请求
from fastapi import FastAPI
from time import sleep
from time import time
import requests
import asyncio
app = FastAPI()
URL= "http://httpbin.org/uuid"
# @app.get("/")
async def main():
r = requests.get(URL)
# print(r.text)
return r.text
async def task():
tasks = [main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main(),main()]
# print(tasks)
# input("stop")
result = await asyncio.gather(*tasks)
print (result)
@app.get('/')
def f():
start = time()
asyncio.run(task())
print("time: ",time()-start)
Run Code Online (Sandbox Code Playgroud)
我将 FastAPI 与 Asyncio 结合使用,以实现大约 3 秒或更短的最短时间,但使用上述方法我得到的总时间为 66 秒,超过一分钟。我还想保留main用于附加操作的功能r.text。我知道要实现如此短的时间,需要并发性,但我不确定我在这里犯了什么错误。
fastapi ×10
python ×6
sqlalchemy ×2
api ×1
async-await ×1
database ×1
django ×1
httprequest ×1
json ×1
pydantic ×1
scikit-learn ×1
starlette ×1