标签: fastapi

我有以下代码：

import time
from fastapi import FastAPI, Request
    
app = FastAPI()
    
@app.get("/ping")
async def ping(request: Request):
        print("Hello")
        time.sleep(5)
        print("bye")
        return {"ping": "pong!"}

如果我在本地主机上运行我的代码 - 例如http://localhost:8501/ping- 在同一浏览器窗口的不同选项卡中，我得到：

Hello
bye
Hello
bye

代替：

Hello
Hello
bye
bye

我已经阅读过有关使用的内容httpx，但仍然无法实现真正​​的并行化。有什么问题？

编辑：

我发现了问题，但不确定为什么会发生这种情况。每当我查询：最后http://localhost:4001/hello/带有“ ”时 - 我都会得到正确的 200 状态响应。/我不懂为什么。

原帖：

每当我向我的应用程序发送查询时，我都会收到 307 重定向。如何让我的应用返回常规状态 200，而不是通过 307 重定向

这是请求输出：

abm                  | INFO:     172.18.0.1:46476 - "POST /hello HTTP/1.1" 307 Temporary Redirect
abm                  | returns the apples data. nothing special here.
abm                  | INFO:     172.18.0.1:46480 - "POST /hello/ HTTP/1.1" 200 OK

pytest 返回：

E       assert 307 == 200
E        +  where 307 = <Response [307]>.status_code

test_main.py:24: AssertionError

在我的根目录：/__init__.py文件：

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
# from .configs import …

我已经按照 FastAPI文档设置了单元测试，但它只涵盖了数据库在测试中保留的情况。

如果我想在每次测试时构建和拆除数据库怎么办？（例如，下面的第二个测试将失败，因为第一个测试后数据库将不再为空）。

我目前通过在每个测试的开始和结束时调用 and （在下面的代码中注释掉）来create_all做到这一点，但这显然不理想（如果测试失败，数据库将永远不会被拆除，影响下一个测试的结果drop_all测试）。

我怎样才能正确地做到这一点？我应该围绕依赖关系创建某种 Pytest 固定装置吗override_get_db？

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from main import app, get_db
from database import Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base.metadata.create_all(bind=engine)

def override_get_db():
    try:
        db = TestingSessionLocal()
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db

client = TestClient(app)

def test_get_todos():
    # Base.metadata.create_all(bind=engine)

    # …

使用 python 模块fastAPI，我不知道如何返回图像。在烧瓶中，我会做这样的事情：

@app.route("/vector_image", methods=["POST"])
def image_endpoint():
    # img = ... # Create the image here
    return Response(img, mimetype="image/png")

这个模块中对应的调用是什么？

在我的项目文件夹中，我有一个基本index.html文件加上静态文件（js、css）以及我的main.py：

from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi import Request

app = FastAPI()

templates = Jinja2Templates(directory="/")
app.mount("/", StaticFiles(directory="/"))

@app.get("/")
def serve_home(request: Request):
    return templates.TemplateResponse("index.html", context= {"request": request}) 

我怎样才能让fastapi在这里工作？我只想index.html在本地主机上提供我的静态文件。没有staticortemplates文件夹有问题吗？

设置：

# Pydantic Models

class TMDB_Category(BaseModel):
    name: str = Field(alias="strCategory")
    description: str = Field(alias="strCategoryDescription")


class TMDB_GetCategoriesResponse(BaseModel):
    categories: list[TMDB_Category]


@router.get(path="category", response_model=TMDB_GetCategoriesResponse)
async def get_all_categories():
    async with httpx.AsyncClient() as client:
        response = await client.get(Endpoint.GET_CATEGORIES)
        return TMDB_GetCategoriesResponse.parse_obj(response.json())

问题：
创建响应时使用别名，我想避免它。我只需要这个别名来正确映射传入数据，但在返回响应时，我想使用实际的字段名称。

实际响应：

{
  "categories": [
    {
      "strCategory": "Beef",
      "strCategoryDescription": "Beef is ..."
    },
    {
      "strCategory": "Chicken",
      "strCategoryDescription": "Chicken is ..."
    }
}

预期回应：

{
  "categories": [
    {
      "name": "Beef",
      "description": "Beef is ..."
    },
    {
      "name": "Chicken",
      "description": "Chicken is ..."
    }
}

这是我的数据库架构。

我这样定义我的架构：

从 pydantic 导入 BaseModel

class Userattribute(BaseModel):\n    name: str\n    value: str\n    user_id: str\n    id: str\n

这是我的模型：

class Userattribute(Base):\n    __tablename__ = "user_attribute"\n\n    name = Column(String)\n    value = Column(String)\n    user_id = Column(String)\n    id = Column(String, primary_key=True, index=True)\n

在 crud.py 中我定义了一个get_attributes方法。

def get_attributes(db: Session, skip: int = 0, limit: int = 100):\n    return db.query(models.Userattribute).offset(skip).limit(limit).all()\n

这是我的GET终点：

@app.get("/attributes/", response_model=List[schemas.Userattribute])\ndef read_attributes(skip: int = 0, limit: int = 100, db: Session = …

我尝试使用 uvicorn 网络服务器运行 FastAPI，但它引发了错误。

我运行这个命令，

uvicorn api:app --reload --host 0.0.0.0

但是终端有错误。

Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Started reloader process [23445]
Error loading ASGI app. Could not import module "api".
Stopping reloader process [23445]

我真的很感激任何建议或建议

我想使用 fastAPI 从函数内的 API 中检索特定标头，但我找不到解决方案。

在烧瓶中很简单：request.headers['your-header-name']

为什么 fastAPI 做这么简单的事情却这么复杂？

有人知道检索标头的解决方案吗？谢谢 ：）

装饰者：

def token_required(f):
    @wraps(f)
    def decorator(*args, **kwargs):
        CONFIG = settings.read_config()
        token = None
        headers = Request.headers
        if "Authorization" in headers:
            auth_header = Request.headers
            token = auth_header
        elif not token:
            return {"Error": "Token is missing or incorrect header name"}, 401

        try:
            public_key = CONFIG["APPLICATION"]["PUBLIC_KEY"]
            claim = jwt.decode(token, public_key)
            claim.validate()
        except UnicodeDecodeError as err:
            return {"Error": f"An error occurred -> {err} check your token"}, 401

        return f(*args, **kwargs)

    return decorator …

我有一个简单的 API 函数，如下所示，

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def read_root():
    return {"Hello": "World"}

我正在使用uvicorn命令启动服务器，

uvicorn main:app

由于我们没有直接调用任何 python 文件，因此无法uvicorn从 Pycharm调用命令。

那么，如何使用 Pycharm 运行 fast-api 服务器？