كيفية إنشاء نقطة نهاية FastAPI التي يمكنها قبول نموذج أو نص JSON؟

في FastAPI، يمكنك إنشاء نقطة نهاية يمكنها قبول نموذج أو نص JSON باستخدام أساليب مختلفة. فيما يلي بعض الخيارات:

الخيار 1: استخدام وظيفة التبعية

يتضمن هذا الخيار إنشاء وظيفة تبعية تتحقق من قيمة رأس طلب نوع المحتوى وتوزع النص باستخدام أساليب Starlette ، وفقًا لذلك.

from fastapi import FastAPI, Depends, Request
from starlette.datastructures import FormData

app = FastAPI()

async def get_body(request: Request):
    content_type = request.headers.get('Content-Type')
    if content_type is None:
        raise HTTPException(status_code=400, detail='No Content-Type provided!')
    elif content_type == 'application/json':
        return await request.json()
    elif (content_type == 'application/x-www-form-urlencoded' or
          content_type.startswith('multipart/form-data')):
        try:
            return await request.form()
        except Exception:
            raise HTTPException(status_code=400, detail='Invalid Form data')
    else:
        raise HTTPException(status_code=400, detail='Content-Type not supported!')

@app.post('/')
def main(body = Depends(get_body)):
    if isinstance(body, dict):  # if JSON data received
        return body
    elif isinstance(body, FormData):  # if Form/File data received
        msg = body.get('msg')
        items = body.getlist('items')
        return msg

الخيار 2: تحديد نقاط نهاية منفصلة

الخيار الآخر هو أن يكون لديك نقطة نهاية واحدة، وتحديد الملف (الملفات) و/أو معلمات بيانات النموذج على أنها اختيارية. إذا تم تمرير قيم إلى أي من المعلمات، فهذا يعني أن الطلب كان إما application/x-www-form-urlencoded أو multipart/form-data. بخلاف ذلك، فمن المحتمل أن يكون طلب JSON.

from fastapi import FastAPI, UploadFile, File, Form
from typing import Optional, List

app = FastAPI()

@app.post('/')
async def submit(items: Optional[List[str]] = Form(None),
                    files: Optional[List[UploadFile]] = File(None)):
    # if File(s) and/or form-data were received
    if items or files:
        filenames = None
        if files:
            filenames = [f.filename for f in files]
        return {'File(s)/form-data': {'items': items, 'filenames': filenames}}
    else:  # check if JSON data were received
        data = await request.json()
        return {'JSON': data}

الخيار 3: استخدام برنامج وسيط

يمكنك أيضًا استخدام برنامج وسيط للتحقق من الطلب الوارد وإعادة توجيهه إلى نقطة النهاية /submitJSON أو /submitForm، اعتمادًا على على نوع محتوى الطلب.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.middleware("http")
async def some_middleware(request: Request, call_next):
    if request.url.path == '/':
        content_type = request.headers.get('Content-Type')
        if content_type is None:
            return JSONResponse(
                content={'detail': 'No Content-Type provided!'}, status_code=400)
        elif content_type == 'application/json':
            request.scope['path'] = '/submitJSON'
        elif (content_type == 'application/x-www-form-urlencoded' or
              content_type.startswith('multipart/form-data')):
            request.scope['path'] = '/submitForm'
        else:
            return JSONResponse(
                content={'detail': 'Content-Type not supported!'}, status_code=400)

    return await call_next(request)

@app.post('/')
def main():
    return

@app.post('/submitJSON')
def submit_json(item: Item):
    return item

@app.post('/submitForm')
def submit_form(msg: str = Form(...), items: List[str] = Form(...),
                    files: Optional[List[UploadFile]] = File(None)):
    return msg

اختبار الخيارات

يمكنك اختبار الخيارات المذكورة أعلاه باستخدام مكتبة طلبات بايثون:

import requests

url = 'http://127.0.0.1:8000/'
files = [('files', open('a.txt', 'rb')), ('files', open('b.txt', 'rb'))]
payload ={'items': ['foo', 'bar'], 'msg': 'Hello!'}
 
# Send Form data and files
r = requests.post(url, data=payload, files=files)  
print(r.text)

# Send Form data only
r = requests.post(url, data=payload)              
print(r.text)

# Send JSON data
r = requests.post(url, json=payload)              
print(r.text)