标签: fastapi

我正在使用 Tortoise-ORM 作为其 orm 来使用 FastAPI，并遇到了问题。具体来说，我无法返回模型中的关系。

这是我的应用程序结构。结构的灵感来自 Django 的应用程序结构。

.\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Dockerfile\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 LICENSE\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Pipfile\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Pipfile.lock\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 README.md\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 app\n\xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 contacts\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 main.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 models.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 routers.py\n\xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 main.py\n\xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 resources\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 constants.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 core_model.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 database.py\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 middlewares.py\n\xe2\x94\x82   \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 users\n\xe2\x94\x82       \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82       \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 main.py\n\xe2\x94\x82       \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 models.py\n\xe2\x94\x82       \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 routers.py\n\xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 docker-compose.yml\n

数据库连接的设置app/resources/database.py如下；

.\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Dockerfile\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 LICENSE\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Pipfile\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Pipfile.lock\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 README.md\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 app\n\xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 contacts\n\xe2\x94\x82   \xe2\x94\x82   \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 __init__.py\n\xe2\x94\x82   \xe2\x94\x82 …

我想知道是否有某种方法可以让我轻松处理输入参数并将它们限制为 FASTAPI 中的多个值。

例如，如果我在这里有一个 hello-world 处理程序：

from fastapi import FastAPI

app = FastAPI()

@app.get(/)
async def root(name:str):
    return {"user_name_is": name}

而我想要实现的是，让用户只能输入以下名称之一作为参数 [ Bob, Jack] ，其他名称都是非法的。

编写一些进一步的检查代码即可达到预期结果并不复杂：

from fastapi import FastAPI

app = FastAPI()

@app.get(/)
async def root(name:str):
    if name in ['Bob' , 'Jack']:
         return {"user_name_is": name}
    else:
         raise HTTPException(status_code=403)

然而，编写代码仍然不够容易，尤其是当需要处理大量输入参数时。我想知道是否有一种方法可以使用类型提示和 pydantic 来达到相同的结果？

在文档中没有找到太多信息，需要帮助，谢谢。

=======

顺便说一句，如果我也有机会需要获取输入参数列表，有什么方法可以检查它们，就像下面的代码一样？

from fastapi import FastAPI
from typing import List

app = FastAPI()

@app.get(/)
async def root(names:List[str]):
    for name in names:
        if name not in …

与 中一样ExpressJS，app.get("/*")适用于所有路线。

我的代码 -

from typing import Optional
from fastapi import FastAPI

app = FastAPI()


@app.get('/*')
def user_lost():
    return "Sorry You Are Lost !"

我尝试过，但网页结果显示{"detail":"Not Found"}
How can I do the same in FastApi?

我创建了一个登录路径，在其中发布表单数据并设置 cookie。设置 cookie 后，我重定向到“/main”，在那里我得到{detail:"Method Not Allowed"}响应。

@app.post("/login")
async def login(request:Request):
     response = RedirectResponse(url="/main")
     response.set_cookie(key="cookie",value="key-value")
     return response

@app.get("/main")
async def root(request:Request, cookie: Optional[str] = Cookie(None)):
     if cookie:
        answer = "set to %s" % cookie
     else:
          answer = "not set"

     return {"value": answer}

我进一步检查了控制台，发现在重定向期间向“/main”发出了 POST 请求，从而导致了错误。当我将其更改为app.post("/main")它时，效果很好。我该如何避免这个错误？我不想每次都发出访问“/main”的发布请求。提前致谢。

我想通过 Alembic 进行迁移，但有些东西不起作用。我不明白我到底做错了什么。

我的蒸馏器 .env

from logging.config import fileConfig

from sqlalchemy import create_engine
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from db import ORMAR_DATABASE_URL
from alembic import context
import sys, os

sys.path.append(os.getcwd())
config = context.config

fileConfig(config.config_file_name)

from db import Base
target_metadata = Base.metadata
URL = "postgresql://admin:admin@localhost/fa_naesmi"



def run_migrations_offline():


context.configure(
    url=URL,
    target_metadata=target_metadata,
    literal_binds=True,
    dialect_opts={"paramstyle": "named"},
    user_module_prefix='sa.'
)

with context.begin_transaction():
    context.run_migrations()


def run_migrations_online():
connectable = create_engine(URL)

with connectable.connect() as connection:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        user_module_prefix='sa.'
    )

    with context.begin_transaction():
        context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline() …

我如何为类似的 url/路由器创建别名（调用相同的处理程序），例如https://myapi/users/5和https://myapi/users/me（我的 id 放置在令牌中，它是 5）。

    @router.get("/{employee_id}}", status_code=200, response_model=schemas.EmployeeOut)\n    async def get_employee(\n            employee_id: int = Path(default=..., ge=0, description=\'Id \xd0\xbf\xd0\xbe\xd0\xbb\xd1\x83\xd1\x87\xd0\xb0\xd0\xb5\xd0\xbc\xd0\xbe\xd0\xb3\xd0\xbe \xd1\x81\xd0\xbe\xd1\x82\xd1\x80\xd1\x83\xd0\xb4\xd0\xbd\xd0\xb8\xd0\xba\xd0\xb0.\', example=8),\n            token: str = Depends(config.oauth2_scheme),\n            postgres_session: AsyncSession = Depends(database.get_db)):\n        try:\n            token_payload = services.get_token_payload(token=token)\n            if token_payload[\'role\'] in (config.OPERATOR_ROLE, config.OPERATOR_ROLE, config.CLINIC_ROLE):\n                return (await postgres_session.execute(statement=models.employees.select().where(\n                    models.employees.c.id == employee_id))).fetchone()\n            else:\n                raise config.known_errors[\'forbidden\']\n        except Exception as error:\n            services.error_handler(error)\n    \n    # Just as example!\n    @router.get("/me}", status_code=200, response_model=List[schemas.TicketOut])\n    async def get_me(\n            token: str = Depends(config.oauth2_scheme)):\n        token_payload = services.get_token_payload(token=token)\n        get_employee(employee_id=token_payload[\'sub\'])\n

这些函数几乎相同，唯一的区别是在第二个函数中没有路径参数employee_id，但它无论如何都存在于令牌中。 …

我正在使用 FastAPI 制作简单的 CRUD API，我想要做的是在创建新项目时生成唯一的随机数（其他字段是应由用户填写的地址和名称）。我怎样才能做到这一点？

我的代码片段包含类和 POST 函数。

app = FastAPI()

userdb = []

class User(BaseModel):
    id: int
    address: str
    name: str

@app.post("/users")
def add_user(user: User):
    userdb.append(users.dict())
    return userdb[-1]

我正在尝试将数据插入 mysql 服务器，当我尝试将数据添加到 SQLalchemy 会话时，我收到错误“生成器对象没有添加属性”

db=get_db()
temp = schema.User(**filtered_dict)
insert_data=models.User(**temp.dict())
db = get_db()
db.add(insert_data)
db.commit()
db.refresh()

会话生成器：

def get_db():
 
    db_session = sessionlocal()
    try:
        yield db_session
    finally:
        db_session.close()

会话创建

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://root:password@localhost:3305/line_api"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL
)
sessionlocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


Base = declarative_base()

请帮忙...

我有一个 Pydantic 模型如下

class Student(BaseModel):
    name:str
    age:int

通过此设置，我希望获得如下 OpenAPI 架构：

那么，如何使用 Pydantic 模型来获取 FastAPI 中的 from 查询参数呢？

我在 VirtualBox 上配置并配置了一台具有 2048 MB RAM 的 Fedora 34 虚拟机，以便在localhost:7070. 完整的应用程序源代码和依赖代码以及说明位于此处。以下是我可以制作的最小的可重现示例。

main.py

import os, pathlib

import fastapi as fast
import aiofiles

        
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
RESULTS_DIR = pathlib.Path('/'.join((ROOT_DIR, 'results')))

    
app = fast.FastAPI()

    
@app.post('/api')
async def upload(
    request: fast.Request, 
    file: fast.UploadFile = fast.File(...),
    filedir: str = ''):
        
    dest = RESULTS_DIR.joinpath(filedir, file.filename)
    dest.parent.mkdir(parents=True, exist_ok=True)

    async with aiofiles.open(dest, 'wb') as buffer:
        await file.seek(0)
        contents = await file.read()
        await buffer.write(contents)

    return f'localhost:7070/{dest.parent.name}/{dest.name}'

start.sh服务器应用程序

#! /bin/bash
uvicorn --host "0.0.0.0" --log-level debug …