如何在 FastAPI 中使用来自 POST 的查询参数?

How do I consume query parameters from POST in FastAPI?

我想在 FastAPI 中编写一个路由来使用 POST 请求。我有以下 URL 示例:

https://URL.com/api/FlexfoneCall/outgoing?AccountId=1234&TimeStamp=2021-05-20+08%3a30%3a56&UniqueCallId=SIP%2f%2b4512345678-0000e4a81463430317&EmployeeLocalNumber=200&ANumber=12345678&BNumber=87654321&PhoneLocalNumber=123456

不过,我只习惯按请求消费body。如何从上述示例中“获取”路径参数数据?

编辑:

我正在尝试编写一个从外部服务接收上述 URL 的服务。我尝试执行以下操作,因为我认为既然它是 POST,我的路线应该如下所示:

@app.post('/callout', response_model=CallIn)
async def create_call_out(callout: CallOut, db: Session = Depends(get_db)):
    db_write_call_out = write_call_out(db, callout)

    return db_write_call_out

使用以下 Pydantic 模型:

class CallOut(BaseModel):
    accountid: str = None
    timestamp: str = None
    uniquecallid: str = None
    employeelocalnumber: str = None
    anumber: str = None
    bnumber: str = None
    phonelocalnumber: str = None

    class Config:
        orm_mode = True

和增删改查功能:

def write_call_out(db: Session, callout: CallOut):
    db_write_call_out = DBCallOut(**callout.dict())
    db.add(db_write_call_out)
    db.commit()
    db.refresh(db_write_call_out)

    return db_write_call_out

SQL用于将查询参数写入 SQL 数据库的 Alchemy ORM 模型:

class DBCallOut(Base):
    __tablename__ = "call_out"

    accountid = Column('AccountId', String(250))
    uniquecallid = Column('UniqueCallId', String(250))
    employeelocalnumber = Column('EmployeeLocalNumber', String(250))
    anumber = Column('ANumber', String(250))
    timestamp = Column('TimeStamp', String(250))
    phonelocalnumber = Column('PhoneLocalNumber', String(250))
    bnumber = Column('BNumber', String(250))
    ID = Column(Integer,  primary_key=True,  index=True, autoincrement=True)

但我一直收到 422 错误。

编辑 2:

最终将我的路线更改为以下内容:

@app.get('/call')
async def outgoing(AccountId: str,
                   TimeStamp: str,
                   UniqueCallId: str,
                   EmployeeLocalNumber: str,
                   ANumber: str,
                   BNumber: str,
                   PhoneLocalNumber: str, db: Session = Depends(get_db)):
    callout = {"AccountId": AccountId,
                "TimeStamp": TimeStamp,
                "UniqueCallId": UniqueCallId,
                "EmployeeLocalNumber": EmployeeLocalNumber,
                "ANumber": ANumber,
                "BNumber": BNumber,
                "PhoneLocalNumber": PhoneLocalNumber}
    db_write_call_out = write_call_out(db, callout)

    return db_write_call_out

这是可行的,但有点笨拙。

我不确定你是想使用 Path 参数(url 路径的一部分,如 google.co/youtube/videos/1)还是想使用查询参数(以路径命名的变量url,如 google.co?video=1) 但无论哪种方式,您只需将路径参数声明为方法装饰 (@app(params)) 的参数,对于查询参数,您在方法参数中声明它们 ( def get_video(params))。 所以对于给定的 link,如果你想获取查询参数,你会这样做:

@app.post("outgoing/")
async def do_something(
      AccountId: int, 
      TimeStamp: str, 
      UniqueCallId: str, 
      EmployeeLocalNumber: str,
      ANumber: int,
      BNumber: int,
      PhoneLocalNumber: int
    ):
    # do some stuff
    save_to_db(owner=AccountId,time=TimeStamp,contact=PhoneLocalNumber)
    

并获取路径变量(坚持我上面给出的例子)你这样做:

@app.post("video/{video_id}")
async def do_something():
    # do something
    save_to_db(video_id)
    

无论您选择实施哪个,都将从请求 url 客户端获取参数值(通过路径参数在路径中的位置,或从查询参数中的同名查询参数)发送并将它们放入您声明的变量名中。然后你像示例所示一样正常使用它们。

文档对新手非常友好,因此查询参数请参考 here, and for path parameters refer here

作为旁注,对于更多 pythonic 代码,我建议您使用 snake case 来命名您的变量