标签: fastapi

在python中，使用Fast API，我有一个在打印时显示的str（这是一个例子，真正的str更复杂）：

[1592494390, 'test', -0.2761097089544078, -0.0852381808812182, -0.101153, nan]

我想使用 Fast API 作为 JSON 数组返回它。

使用 JSONResponse

def get_json(dataset: str, timeseries: str):

    test = "[1592494390, 'test', -0.2761097089544078, -0.0852381808812182, -0.101153, nan]"
    print(test)
    return JSONResponse(content=test)

打印结果如预期所示：

[1592494390, 'test', -0.2761097089544078, -0.0852381808812182, -0.101153, nan]

但是 API 在调用时的答案是：

"[1592494390, 'test', -0.2761097089544078, -0.0852381808812182, -0.101153, nan]"

所以我的str再次被连载，我不知道如何绕过它。

使用响应：

Fast API 的文档中有一个页面描述了如何直接返回响应（https://fastapi.tiangolo.com/advanced/response-directly/），其中写道：

当您直接返回响应时，其数据不会被验证、转换（序列化），也不会自动记录。

但使用这种方法会导致错误：

def get_json(dataset: str, timeseries: str):

    test = "[1592494390, 'test', -0.2761097089544078, -0.0852381808812182, -0.101153, nan]"
    print(test)
    return Response(content=test, media_type="application/json")

> line 53, in …

在 Flask 中，可以将任意函数分配给路由，功能如下：

from flask import Flask
app = Flask()

def say_hello():
    return "Hello"

app.add_url_rule('/hello', 'say_hello', say_hello)

等于（使用装饰器）：

@app.route("/hello")
def say_hello():
    return "Hello"

add_url_ruleFastAPI中有没有这么简单又实用的方式（ ）？

我正在这个项目上学习 FastAPI 并遇到了这个问题 - 任何请求，无论 GET 或 POST 都会抛出一个错误：值不是有效的字典（type=type_error.dict）。我的代码有什么问题。我的模型.py

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    telegram_id = Column(Integer, unique=True, index=True)
    username = Column(String(50))
    pet_name = Column(String(50))
    language_code = Column(String(5))

    sent_items = relationship("Log", back_populates="recipient")

class Log(Base):
    __tablename__ = "sent_log"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"))
    recipient = relationship("User", back_populates="sent_items")

    article_id = Column(Integer, ForeignKey("articles.id"))
    article = relationship("Article", back_populates="items")

class Article(Base):
    __tablename__ = "articles"

    id = Column(Integer, primary_key=True, index=True)
    text = Column(String(1024))
    image_url = Column(String(500), index=True) …

在我的 Fast API 应用程序中，我有这个 pydantic 模型

class UserInArticleView(BaseModel):
    """What fields will be in nested sent_to_user list."""

    telegram_id: int

    class Config:
        """Enable ORM mode."""

        orm_mode = True

class ArticleBase(BaseModel):
    id: int
    text: str = Field(..., min_length=50, max_length=1024)
    image_url: HttpUrl = Field(..., title="Image URL")
    language_code: str = Field("ru", max_length=3, min_length=2)
    sent_to_user: List[UserInArticleView] = []

    class Config:
        orm_mode = True

响应是

[
  {
    "id": 1,
    "text": "Some text",
    "image_url": "http://test.tt/",
    "language_code": "ru",
    "sent_to_user": [
      {
        "telegram_id": 444444444
      },
      {
        "telegram_id": 111111111
      }
    ] …

我正在尝试创建一个简单的可插入 FastAPI 应用程序，其中插件可以添加或不添加 API 端点

这是我的文件夹结构：

服务器.py

import importlib
import pkgutil
from pathlib import Path

import uvicorn
from fastapi import FastAPI

PLUGINS_PATH = Path(__file__).parent.joinpath("plugins")
app = FastAPI()


def import_module(module_name):
    """Imports a module by it's name from plugins folder."""
    module = f"plugins.{module_name}"
    return importlib.import_module(module, ".")


def load_plugins() -> list:
    """Import plugins from plugins folder."""
    loaded_apps = []
    for _, application, _ in pkgutil.iter_modules([str(PLUGINS_PATH)]):
        module = import_module(application)
        print(
            f"Loaded app: {module.__meta__['plugin_name']} -- version: {module.__meta__['version']}"
        )
        loaded_apps.append(module)
    return loaded_apps


@app.get("/")
def main():
    return …

我对 Python 很陌生。

我正在使用fastapi和制作一个简单的网络服务器uvicorn。

当我构建 Docker 并运行它时，我得到以下内容：

INFO:     Started server process [1]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:5000 (Press CTRL+C to quit)

我的代码是：

# Setup FastAPI server
import uvicorn
from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every
from fastapi.middleware.cors import CORSMiddleware
from dotenv import load_dotenv
import os
from pymongo import MongoClient
import logging

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

load_dotenv('env')

# Database
CONNECTION_URL = os.environ['mongoConnectionURL']
DATABASE_NAME = os.environ['mongoDatabaseName']
NEWS_COLLECTION = …

我转而使用 TortoiseORM 中的 SQLAlchemy，并认为我应该研究 Alembic 来处理其迁移。编辑env.py和alembic.ini文件后，我仍然无法让 alembic 生成任何迁移。

该错误sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/14/xd2s) sys:1: RuntimeWarning: coroutine 'connect' was never awaited是不言自明的，但我不知道到底要更改什么。

我正在遵循FastAPI-Users文档中的指示，但完全迷失了。

我尝试过的：

• 设置run_migrations_offline()和run_migrations_online()作为async
• 使用asyncio.run()to 这样我就可以运行它们

模型.py

import os
from typing import AsyncGenerator
from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from …

I have created a simple API using FastAPI, and I am trying to pass a URL to a FastAPI route as an arbitrary path parameter.

from fastapi import FastAPI
app = FastAPI()
@app.post("/{path}")
def pred_image(path:str):
    print("path",path)
    return {'path':path}

When I test it, it doesn't work and throws an error. I am testing it this way:

http://127.0.0.1:8000/https://raw.githubusercontent.com/ultralytics/yolov5/master/data/images/zidane.jpg

如何在 FastAPI Swagger 自动文档中设置 API 方法的排序顺序？我希望所有方法按类型分组（GET、POST、PUT、DELETE）。

这个答案展示了如何在 Java 中做到这一点。我怎样才能用Python做到这一点？

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def list_all_components():
    pass

@app.get("/{component_id}")
def get_component(component_id: int):
    pass

@app.post("/")
def create_component():
    pass

@app.put("/{component_id}")
def update_component(component_id: int):
    pass

@app.delete("/{component_id}")
def delete_component(component_id: int):
    pass

我是一名 Java Spring boot 开发人员，我开发 3 层 CRUD 应用程序。我和一个似乎对这个主题很了解的人交谈过，但我没有得到他的联系方式。他提倡使用 Python 的 FastApi，因为它的水平扩展性比 Spring boot 更好。他提到的原因之一是FastApi是单线程的。当线程遇到数据库查找（或其他可以异步完成的工作）时，它会选择其他工作，以便稍后在数据库结果传入时返回到当前工作。在 Java 中，当有许多请求待处理时，线程池可能会耗尽。

我不能百分百理解这个推理。让我扮演魔鬼代言人。当Python程序遇到异步调用时，它必须以某种方式将程序指针存储在某个地方，以记住稍后需要在哪里继续。我知道存储程序指针的地方根本不是线程，但我必须给它起个名字，所以我们称它为“逻辑线程”。在 Python 中，您可以有许多正在等待的逻辑线程。在 Java 中，您可以拥有一个线程池，其中有许多正在等待的实际线程。对我来说，唯一的区别似乎是 Java 的线程是在操作系统级别管理的，而 Python 的“逻辑线程”是由 Python 或 FastApi 管理的。为什么在线程池中等待的实际线程比等待的逻辑线程要昂贵得多？如果我的大多数线程都在等待，为什么我不能增加线程池大小以避免耗尽？