我试图返回 Company 类型的对象列表，仅包括“已批准”的对象，并且具有或多或少的属性，具体取决于请求列表的用户是超级用户还是普通用户。到目前为止，这是我的代码：

@router.get("/", response_model=List[schema.CompanyRegularUsers])
def get_companies(db: Session = Depends(get_db), is_superuser: bool = Depends(check_is_superuser)):
    """
    If SU, also include sensitive data.
    """
    if is_superuser:
        return crud.get_companies_admin(db=db)
    return crud.get_companies_user(db=db)
#

该函数根据请求正确返回对象（即，仅is_approved=True公司如果是常规请求，并且两者is_approved=True都由is_approved=False超级用户请求。问题是，两种情况都使用schema.CompanyRegularUsers，并且我想schema.CompanySuperusers在 SU 发出请求时使用。

我怎样才能实现这个功能？即，有没有办法有条件地设置response_model装饰器函数的属性？

我尝试过使用JSONResponse和调用 Pydantic's schema.CompanySuperusers.from_orm()，但它不适用于公司列表......

我正在尝试学习和使用 AWS Cognito 用户池，并与 Python FastAPI 实现的 API 集成。到目前为止，我正在使用授权代码流，将我的 Cognito 用户池重定向到 FastAPI 上的端点来解决代码挑战。源代码附加在此查询的末尾。

该 API 具有以下端点：

1. 根终端节点 [ / ]：将浏览器重定向到我的 AWS Cognito 用户池的登录页面。
2. 重定向端点 [ /aws_cognito_redirect ]：成功登录用户池后激活。接收来自 cognito 用户池的代码质询。在下面显示的代码中，aws_cognito_redirect终端节点通过将代码质询、redirect_uri、client_id 等发送到 AWS Cognito 用户池oauth2/token终端节点来解决代码质询。我可以在控制台日志输出中看到身份、访问和刷新令牌已成功检索。

FastAPI 另外还有一些受保护的端点，这些端点将从 Web 应用程序中调用。此外，还将有一个 Web 表单与端点进行交互。

在此阶段，我可以使用 FastAPI jinja2 模板来实现和托管 Web 表单。如果我选择此选项，大概我可以让/aws_cognito_redirect端点在仅 HTTP 会话 cookie 中返回令牌。这样，每个后续的客户端请求都会自动包含 cookie，而不会在浏览器本地存储中暴露任何令牌。我知道我必须使用此选项来处理 XSRF/CSRF。

或者，我可以使用 Angular/React 来实现前端。据推测，推荐的做法似乎是我必须将授权流程重新配置为使用 PKCE 的身份验证代码？在这种情况下，Angular/React Web 客户端将直接与 AWS Cognito 通信，以检索将转发到 FastAPI 端点的令牌。这些令牌将存储在浏览器的本地存储中，然后在每个后续请求的授权标头中发送。我知道这种方法容易受到 XSS 攻击。

在这两者中，考虑到我的要求，我认为我倾向于使用 jinja2 模板在 FastAPI 上托管 Web 应用程序，并在成功登录时返回仅 …

无法理解，即使我删除所有内部函数并只打印一些内容仍然出现此错误，但是当我使用 fastapi 文档并尝试使用它进行签名时，它可以工作。

@auth_router.post('/signin')
async def sign_in(username: str = Form(...), password: str = Form(...)) -> dict:
    user = await authenticate_user(username, password)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, 
            detail='Invalid username or password',
        )

    user_obj = await User_Pydantic.from_tortoise_orm(user)
    user_token = await generate_token(user_obj)

    return {
        'access_token': user_token,
        'token_type': 'bearer',
    }

在我使用 OAuth2PasswordRequestForm 之前，当出现 422 错误时，请尝试其他方法。

我的模型是 tortoise orm，当需要时我将其转换为 pydantic 模型，在文档中一切正常。

JS

handleEvent(signinform, 'submit', e => {
    e.preventDefault();
    if(!isEmpty(signinform)){

        signInUsername = getElement('input[name="username"]', signinform).value;
        signInPassword = getElement('input[name="password"]', signinform).value;
        recaptchaV3 = getElement('[name="g-recaptcha-response"]').value;

        if(recaptchaV3){
            signInData = new …

我有一个获取函数，它接受多个查询参数，可能如下所示：

def get(
   key: Optional[str] = "key"
   value: Optional[str] = "value"
   param1: Optional[int] = -1
)

我想要做的是，我想将这些参数定义放在一个单独的变量中。可以做这样的事情吗？

param_definition = { # some struct here, or maybe a Model class
   key: Optional[str] = "key"
   value: Optional[str] = "value"
   param1: Optional[int] = -1
}


def get(*params: param_definition):
   ...

这可以做到吗？如果没有，是否有类似且更易于维护的事情可以在这里完成？

我是这部分编程的新手，我有几个问题。首先是我的项目。一侧有一个 Flutter 应用程序，另一侧有一个包含数据的 MS SQL Server。从逻辑上讲，我的设备上需要这些数据。我读到最好的方法是使用 FastAPI，它简单且性能良好，但我不确定安全性。我读过一些有关 OAuth2 的内容，但它看起来太多了，因为只有一个用户有权使用该数据（服务器所有者）。是否可以只使用一个简单的 api key 作为参数？像这样的东西...

from fastapi import FastAPI
from SqlServerRequest import SqlServerRequest

app = FastAPI()


@app.get("/openOrders/{key}")
async def openOrders(key):
    if key == "myverysecurekey":
         return "SQLDATA"
    else
         return "Wrong key"

这种方式可行，但我不确定安全性你会说什么？

我正在使用 FastAPI 构建数据 API。我希望客户发布 2 个包含 24 个浮点数的列表，稍后我会将其保存到数据库中。

当我尝试创建 Pydantic 模型时：

from pydantic import BaseModel

class Prices(BaseModel):
    buying_price: list(float)=[]
    selling_price: list(float)=[]

我收到以下错误：

File "c:/Users/Amin y Lubna/FastAPI-InfluxDB/test.py", line 3, in <module>
    class Prices(BaseModel):
File "c:/Users/Amin y Lubna/FastAPI-InfluxDB/test.py", line 4, in Prices
    buying_price: list(float)=[]
TypeError: 'type' object is not iterable

尽管该错误是不言自明的，但我不明白。

然后，查看文档，我发现了以下方法：

from pydantic import BaseModel
from typing import List

class Prices(BaseModel):
    buying_price: List(float)=[]
    selling_price: List(float)=[]

但我收到以下错误。

File "c:/Users/Amin y Lubna/FastAPI-InfluxDB/test.py", line 4, in <module>
    class Prices(BaseModel):
File "c:/Users/Amin y Lubna/FastAPI-InfluxDB/test.py", line …

fastapi nginx 模板

源代码

它是快速的 api 和 nginx 模板，但目前它不能按我的预期工作。\n当你卷曲本地主机时，它响应 502 bad gateway instaed of hello world 这是在 main.py 中定义的...

docker 日志 web 显示connect() failed (111: Connection refused) while connecting to upstream, client: 172.27.0.1, server:在 nginx 容器中。

但为什么 ？？你如何修复这个错误？

文件树

.\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Dockerfile\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 README.md\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 docker-compose.yml\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 main.py\n\xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 nginx\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x9c\xe2\x94\x80\xe2\x94\x80 Dockerfile\n\xe2\x94\x82\xc2\xa0\xc2\xa0 \xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 nginx.conf\n\xe2\x94\x94\xe2\x94\x80\xe2\x94\x80 requirements.txt\n

构建并启动

\n

1. docker-compose up -d --build

   \n
\n

2. docker logs ps -a

   \n
\n

CONTAINER ID   IMAGE                         COMMAND                  CREATED …

我目前正在开发一个小项目，其中涉及创建一个fastapi允许用户上传文件的服务器jar文件的服务器。

基本上我有这条路线：

@app.post("/upload")
async def upload(jar_file: UploadFile = File(...)):

我真的很想检查并验证该文件是否确实是jar文件。

我可以自己实现它，但我很好奇是否fastapi或任何其他包提供此功能。

我尝试了本教程（https://dev.to/rajshirolkar/fastapi-over-https-for-development-on-windows-2p7d），然后当我进入浏览器但无法使用 https 访问我的本地主机地址时，如果我尝试使用 http，它可以工作，但我需要使用 HTTPS。

这是我尝试过的代码：

import uvicorn

if __name__ == "__main__":
    uvicorn.run("app.api:app",
                 host="localhost",
                 port=8432, 
                 reload=True,
                 ssl_keyfile="./key.pem", 
                 ssl_certfile="./cert.pem")

pydantic中有一个DocumentSchema用FastApi编写的类的嵌套规则如下：

class DocumentSchema(BaseModel):
    clientName: str
    transactionId: str
    documentList: List[SingleDocumentSchema]

和

class SingleDocumentSchema(BaseModel):
    documentInfo: DocumentInfoSchema
    articleList: List[DocumentArticleSchema]

和

class DocumentInfoSchema(BaseModel):
    title: str
    type: str
    referenceId: int
    batchNoList: Optional[List]
    otherData: Optional[Json]

和

class DocumentArticleSchema(BaseModel):
    type: str
    value: int
    accountType: Optional[AccountTypeEnums]
    accountId: Optional[int]
    otherData: Optional[Json]

这是从 Kafka 接收消息并处理它的 python 代码片段：

def process(self) -> bool:
    try:
        DocumentSchema(
            **json.loads(self._message)
        )
        return self._process()

    except ValidationError as e:
        raise UnprocessableEntityException(e, self._topic)
    except ValueError as e:
        raise UnprocessableEntityException(e, self._topic)
    except Exception as e:
        raise UnprocessableEntityException(e, self._topic) …