标签: fastapi

在 FastAPI 框架内：

虽然请求数据当然可以作为参数传递，但我想知道函数是否可以在不传递参数的情况下访问有关当前请求的信息。

免责声明：我不认为全局访问请求数据是一个好的做法，但我有一个用例，能够做到这一点会非常好。

我有一个 FastAPI 应用程序，在本地运行时按预期工作，但是，当我尝试在 Docker 容器中运行时，出现“内部服务器错误”。这是我的应用程序的代码：

from fastapi import FastAPI
from pydantic import BaseModel
import pandas as pd
from fbprophet import Prophet

class Data(BaseModel):
    length: int
    ds: list
    y: list
    model: str
    changepoint: float = 0.5
    daily: bool = False
    weekly: bool = False
    annual: bool = False
    upper: float = None
    lower: float = 0.0
    national_holidays: str = None

app = FastAPI()

@app.post("/predict/")
async def create_item(data: Data):

    # Create df from base model
    df = pd.DataFrame(list(zip(data.ds, data.y)), columns =['ds', 'y']) …

我使用 Fast API 创建 Web 服务。

有以下 sqlAlchemy 模型：

class User(Base):
    __tablename__ = 'user'
    account_name = Column(String, primary_key=True, index=True, unique=True)
    email = Column(String, unique=True, index=True, nullable=False)
    roles = relationship("UserRole", back_populates="users", lazy=False, uselist=True)


class UserRole(Base):
    __tablename__ = 'user_role'
    __table_args__ = (UniqueConstraint('role_name', 'user_name', name='user_role_uc'),)
    role_name = Column(String, ForeignKey('role.name'), primary_key=True)
    user_name = Column(String, ForeignKey('user.account_name'), primary_key=True)
    users = relationship("User", back_populates="roles")

Pydantic 架构如下：

class UserRole(BaseModel):
    role_name: str

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    account_name: str
    email: EmailStr
    roles: List[UserRole] = []

    class Config:
        orm_mode …

GET我是 FastAPI 的新手（从 Flask 迁移），我正在尝试为我的路线创建一个 Pydantic 模型：

from fastapi import APIRouter,Depends
from pydantic import BaseModel
from typing import Optional,List

router = APIRouter()

class SortModel(BaseModel):
    field:    Optional[str]
    directions: List[str]

@router.get("/pydanticmodel")
def get_sort(criteria: SortModel = Depends(SortModel)):
    pass #my code for handling this route.....

当我跑步时 curl -X GET http://localhost:XXXX/pydanticmodel?directions=up&directions=asc&field=id 我得到422 Unprocessable Entity: {"detail":[{"loc":["body"],"msg":"field required","type":"value_error.missing"}]}

但如果我正在改变directions:List[str]->directions: str我正在200 OK接受directions="asc". str对于查询参数有效而List[str]无效的原因是什么？我究竟做错了什么？

谢谢。

我需要创建一个具有路由的 API，该路由能够识别当前用户是否是请求中指示的用户（而且任何身份验证都不应有效）

对于其他路径，我遵循https://fastapi.tiangolo.com/tutorial/security/oauth2-jwt/，一切都可以与带有 JWT 令牌的 Bearer 一起使用，如下所示

user: User = Depends(get_current_active_user)

修改我尝试过的文档提供的方法

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth")

async def get_user_or_none(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
    """
    Return the current active user if is present (using the token Bearer) or None
    """
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            return None
    except JWTError:
        return None
    
    # check user in db
    user = crud.get_user(db, username)
    if user is None:
        return None
    return user …

我正在尝试向我的模型添加一个path类型字段PureWindowsPath。按照此处建议实现自定义验证器后https://github.com/samuelcolvin/pydantic/issues/2089#issuecomment-890018075在尝试访问 SwaggerUI 时出现以下错误：

INFO:     127.0.0.1:7696 - "GET /api/openapi.json HTTP/1.1" 500 Internal Server Error
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "C:\Users\xxx\.virtualenvs\server-J9mXI7Iu\lib\site-packages\uvicorn\protocols\http\httptools_impl.py", line 371, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "C:\Users\xxx\.virtualenvs\server-J9mXI7Iu\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 59, in __call__
    return await self.app(scope, receive, send)
  File "C:\Users\xxx\.virtualenvs\server-J9mXI7Iu\lib\site-packages\fastapi\applications.py", line 208, in __call__
    await super().__call__(scope, receive, send)
  File "C:\Users\xxx\.virtualenvs\server-J9mXI7Iu\lib\site-packages\starlette\applications.py", line 112, in __call__
    await self.middleware_stack(scope, receive, send)
  File "C:\Users\xxx\.virtualenvs\server-J9mXI7Iu\lib\site-packages\starlette\middleware\errors.py", line 181, in __call__
    raise exc …

我想定义一个从文本文件生成的 dict 变量一次，并用它来响应 API 请求。

该变量应该始终可用，直到服务器运行结束。

在下面的示例中：

from fastapi import FastAPI
import uvicorn

app = FastAPI()

def init_data(path):
    print("init call")
    data = {}
    data[1] = "123"
    data[2] = "abc"
    return data

data = init_data('path')

@app.get('/')
def example_method():
    # data is defined
    return {'Data': data[1]}

if __name__ == '__main__':
    uvicorn.run(f'example_trouble:app', host='localhost', port=8000)

我会得到：

init call
init call
INFO:     Started server process [9356]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:8000 (Press CTRL+C to quit)

向 …

我最近将使用 FastApi 编码的 REST API 迁移到新的 SQLAlchemy 1.4+ 异步版本。我的应用程序编译正确，数据库似乎设置得很好。当我尝试执行多个请求时，会出现问题，出现错误，似乎表明同一会话正在用于我的所有请求。我把错误信息放在最后

这是我的代码，我基于 SQLAlchemy 异步文档和此示例

应用程序引擎初始化

from typing import AsyncIterator
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from .notification import Notification
from .devices import Device



from sqlalchemy import MetaData
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import create_engine

from app.core.config import Config
from app.core.constants import logger
import asyncio

engine = create_async_engine(Config.RDS_DB_URL)
metadata = MetaData(engine)

#if not engine.dialect.has_table(engine, C.NOTIFICATIONS):
#Base.metadata.create_all(engine)

async def init_connection():
    async with engine.begin() as conn:
        await …

我正在使用nsidnev/fastapi-realworld-example-app。

我需要将事务逻辑应用到这个项目中。

在一个 API 中，我从存储库调用许多方法，并在许多表中执行更新、插入和删除操作。如果这些操作出现异常，我该如何回滚更改？（或者如果一切正确则提交。）

tl;dr

1. 以下哪个选项是正确的工作流程fastapi？
2. 如何以编程方式测试调用是否真正被阻止（而不是从浏览器手动测试）？是否有对or 的压力测试扩展？uvicornfastapi

我在服务器中有许多端点fastapi（uvicorn目前正在使用），它们对常规同步 Python 代码有长时间的阻塞调用。尽管有文档（https://fastapi.tiangolo.com/async/），我仍然不清楚我是否应该单独使用def或async def混合我的功能。

据我了解，我有三个选择，假设：

def some_long_running_sync_function():
  ...

选项 1 - 始终def仅用于端点

@app.get("route/to/endpoint")
def endpoint_1:
  some_long_running_sync_function()


@app.post("route/to/another/endpoint")
def endpoint_2:
  ...

选项 2 - 始终async def仅使用并在执行器中运行阻塞同步代码

import asyncio


@app.get("route/to/endpoint")
async def endpoint_1:
  loop = asyncio.get_event_loop()
  await loop.run_in_executor(None, some_long_running_sync_function)


@app.post("route/to/another/endpoint")
async def endpoint_2:
  ...

选项 3 - 混合搭配def并async def基于底层调用

import asyncio


@app.get("route/to/endpoint")
def endpoint_1: …