为什么我在记录日志时 FastAPI 不创建日志文件
这是我使用的代码
from fastapi.logger import logger as fastapi_logger
from logging.handlers import RotatingFileHandler
import logging
formatter = logging.Formatter(
"[%(asctime)s.%(msecs)03d] %(levelname)s [%(thread)d] - %(message)s", "%Y-%m-%d %H:%M:%S")
handler = RotatingFileHandler('logfile.log', backupCount=0)
logging.getLogger("fastapi")
fastapi_logger.addHandler(handler)
handler.setFormatter(formatter)
fastapi_logger.info('****************** Starting Server *****************')
与 Flask 不同，文件“logfie.log”不会自动生成。还有其他配置吗？
我有一个简单的FastAPI应用程序，我正在尝试为其创建测试pytest。
我的目标是测试应用程序在出现不同错误时的行为方式。
\n\n我的应用程序中有一个简单的健康检查路线：
\n\nfrom fastapi import APIRouter\n\nrouter = APIRouter()\n\n\n@router.get("/health")\nasync def health():\n return "It\'s working \xe2\x9c\xa8"\n现在在我的 pytest 模块中，我正在尝试修补上述函数，以便它引发不同的错误。\n我正在使用，unittest.mock但我得到了非常奇怪的行为。
import pytest\nfrom unittest import mock\n\nfrom fastapi import HTTPException\nfrom starlette.testclient import TestClient\n\nimport app.api.health\nfrom app.main import app # this is my application (FastAPI instance) with the `router` attached\n\n\n@pytest.fixture()\ndef client():\n with TestClient(app) as test_client:\n yield test_client\n\n\ndef test_simple(client):\n def mock_health_function():\n raise HTTPException(status_code=400, detail=\'gibberish\')\n\n with mock.patch(\'app.api.health.health\', mock_health_function):\n response = client.get(HEALTHCHECK_PATH)\n\n with pytest.raises(HTTPException): # this check passes …我已关注有关 Oauth2 的文档，但它没有描述添加客户端 id 和密钥的过程
https://fastapi.tiangolo.com/advanced/security/oauth2-scopes/
这是做什么的
class UserInDB(User):
hashed_password: str
从原来的例子
如果在字段中传递 None ，我可以在 Pydantic 中设置默认值吗？
我有以下代码，但在我看来，这里的验证器仅适用于模型的初始化，而不适用于其他情况。
我的代码：
class User(BaseModel):
name: Optional[str] = ''
password: Optional[str] = ''
email: EmailStr
@validator('name')
def set_name(cls, name):
return name or 'foo'
遇到的问题：
user = User(name=None, password='some_password', email='user@example.com')
print("Name is ", user.name)
# > 'Name is foo'
user.name = None
print("Name is ", user.name)
# > 'Name is None'
期望输出：
user = User(name='some_name', password='some_password', email='user@example.com')
user.name = None
print("Name is ", user.name)
# > 'Name is foo'
关于如何获得所需输出的任何想法？我认为拥有 getter 和 setter 将有助于解决这个问题。但是，我无法让它们在 Pydantic 模型中工作：
尝试实现 …
我想创建一个 Pydantic 模型，其中有一个列表字段，未初始化的默认值为空列表。有没有一种惯用的方法来做到这一点？
对于 Python 的内置数据类对象，您可以使用field(default_factory=list)，但是在我自己的实验中，这似乎可以防止我的 Pydantic 模型被腌制。一个天真的实现可能是这样的：
from pydantic import BaseModel
class Foo(BaseModel):
defaulted_list_field: Sequence[str] = [] # Bad!
但是我们都知道不要使用像空列表文字这样的可变值作为默认值。
那么给 Pydantic 列表字段一个默认值的正确方法是什么？
我尝试创建 Foo 的子级，每个子级都应该有自己的 uuid。在真实的代码中，不会创建 Foo 的实例，只会创建它的子实例。子项稍后将保存在数据库中，uuid 用于从数据库中检索正确的对象。
在第一个代码片段中，我尝试使用 init 方法，这会导致 AttributeError。我还尝试使用类方法，这会导致丢失子对象中的所有字段。
如果我的第二个片段每个孩子都会获得相同的 uuid，这对我来说很有意义，因为它作为默认值传递。
我可以将 uuid 创建放入子级中，尽管在使用继承时这感觉不对。
有没有更好的方法为每个孩子创建一个uuid？
# foo_init_.py
class Foo(BaseModel):
def __init__(self):
self.id_ = uuid4()
# >>> AttributeError: __fields_set__
# foo_classmethod.py
class Foo(BaseModel):
@classmethod
def __init__(cls):
cls.id_ = uuid4()
# >>> Bar loses id_ fields
from uuid import uuid4, UUID
from pydantic import BaseModel
class Foo(BaseModel):
id_: UUID = uuid4()
class Bar(Foo):
pass
class Spam(Foo):
pass
if __name__ == '__main__':
b1 = Bar()
print(b1.id_) # >>> 73860f46-5606-4912-95d3-4abaa6e1fd2c
b2 = …TestClient我尝试使用FastAPI（基本上是 Scarlett TestClient）测试端点。
响应代码始终为 422 Unprocessable Entity。
这是我当前的代码：
from typing import Dict, Optional
from fastapi import APIRouter
from pydantic import BaseModel
router = APIRouter()
class CreateRequest(BaseModel):
number: int
ttl: Optional[float] = None
@router.post("/create")
async def create_users(body: CreateRequest) -> Dict:
return {
"msg": f"{body.number} Users are created"
}
正如您所看到的，我还将application/json标头传递给客户端以避免潜在的错误。
这是我的测试：
from fastapi.testclient import TestClient
from metusa import app
def test_create_50_users():
client = TestClient(app)
client.headers["Content-Type"] = "application/json"
body = {
"number": 50,
"ttl": 2.0
}
response = client.post('/v1/users/create', data=body) …from fastapi import Depends, FastAPI, HTTPException, Body, Request
from sqlalchemy import create_engine, Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker, relationship
from sqlalchemy.inspection import inspect
from typing import List, Optional
from pydantic import BaseModel
import json
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
app = FastAPI()
# sqlalchemy models
class RootModel(Base):
__tablename__ = "root_table"
id = Column(Integer, primary_key=True, index=True)
someRootText = Column(String)
subData …我是在 DRF 的背景下问这个问题的，但这也可以概括。我们正在维护一个 API 服务，为超过 1K 的实时请求提供服务。在当前场景中，我们正在实时计算 API 调用，即我们正在针对每个 API 调用的用户更新 API 调用计数的数据库列。
但有没有有效的方法呢？例如，在其他地方记录 API 调用，然后在几分钟后更新数据库？或者说我们现在正在做的事情可以吗？大规模 API 服务可以处理多大程度的问题？网上没有针对此问题的具体内容。
我写了一个 fastapi 应用程序。现在我正在考虑部署它，但是我似乎遇到了奇怪的意外性能问题，这似乎取决于我使用 uvicorn 还是 gunicorn。特别是如果我使用 gunicorn，所有代码（甚至标准库纯 python 代码）似乎都会变慢。对于性能调试，我编写了一个小应用程序来演示这一点：
import asyncio, time
from fastapi import FastAPI, Path
from datetime import datetime
app = FastAPI()
@app.get("/delay/{delay1}/{delay2}")
async def get_delay(
delay1: float = Path(..., title="Nonblocking time taken to respond"),
delay2: float = Path(..., title="Blocking time taken to respond"),
):
total_start_time = datetime.now()
times = []
for i in range(100):
start_time = datetime.now()
await asyncio.sleep(delay1)
time.sleep(delay2)
times.append(str(datetime.now()-start_time))
return {"delays":[delay1,delay2],"total_time_taken":str(datetime.now()-total_start_time),"times":times}
使用以下命令运行 fastapi appi：
gunicorn api.performance_test:app -b localhost:8001 -k uvicorn.workers.UvicornWorker --workers 1
get to 的响应体 …