FastAPI / OpenAPI - Callbacks und Responses

Django, Flask, Bottle, WSGI, CGI…
Antworten
Benutzeravatar
Felix92
User
Beiträge: 133
Registriert: Mittwoch 7. November 2018, 17:57

Huhu,
und zwar habe ich gerade folgendes Problem:
Ich bekomme von einer externen API eine URL für den Callback und möchte dort nun per Response auch meine Ergebnisse hinsenden wenn die Bearbeitung fertig ist.
Funktioniert bisher auch alles soweit allerdings kommt der Response nicht bei der Callback-URL an wo liegt der Fehler ?

Also anstelle

Code: Alles auswählen

return {'documentId': doc_id, 'content': str(text), 'errors': errors}
müsste der return an die Callback URL gehen, welche beim Request an den Endpunkt mitgeliefert wird:

Code: Alles auswählen

@app.post("/txt", callbacks=router.routes)
async def ocr_txt(invoice: Invoice, callback_url: Optional[AnyHttpUrl] = None, res_width: int=3500):

Die Docu dazu von FastAPI hatte ich mir schon angesehen allerdings werde ich nicht so ganz schlau daraus.
https://fastapi.tiangolo.com/advanced/o ... callbacks/

Code: Alles auswählen

app = FastAPI()
router = APIRouter()

app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
app.include_router(router)

class Invoice(BaseModel):
    documentId: str
    name: str
    content: str


#class InvoiceEvent(BaseModel):
#    documentId: str
#    content: str
#    errors: list


class InvoiceEventReceived(BaseModel):
    ok: bool


@router.post(
    "{$callback_url}/text/{$request.body.documentId}", response_model=InvoiceEventReceived
)
@router.post(
    "{$callback_url}/pdfa/{$request.body.documentId}", response_model=InvoiceEventReceived
)


#def invoice_notification(body: InvoiceEvent):
#    pass

@app.get("/")
async def root():
    """Entrypoint
    """
    return {"message": "OCR Engine"}

@app.get("/status")
async def status():
    """status time
    """
    version = "v1.5"
    now = datetime.now()
    dt_string = now.strftime(r"%d/%m/%Y %H:%M:%S")
    return {"date and time": dt_string, "version": version}


@app.post("/txt", callbacks=router.routes)
async def ocr_txt(invoice: Invoice, callback_url: Optional[AnyHttpUrl] = None, res_width: int=3500):
    """extract text from pdf or image and return base64 encoded txt 

    Parameters
    ----------
    invoice : Invoice
        invoice form
    callback_url : Optional[AnyHttpUrl], optional
        the client API, by default None
    res_width : int, optional
        the resolution width, by default 3500

    Returns
    -------
    JSONResponse
        document id, base64 content, list of errors
    """
    doc_id = invoice.documentId
    file_name = invoice.name
    file_content = invoice.content
    # fix incorrect base64 padding
    file_content = f"{file_content}{'=' * (len(file_content) % 4)}"

    try:
        try:
            file = base64.b64decode(str(file_content), validate=True)
        except:
            return {'documentId': doc_id, 'content': None, 'errors':{'status_code':444, 'detail':'Internal decoding error'}}
        _, extension = os.path.splitext(file_name)
        text, pdfa, errors = __handling_pdf_and_image(file, extension=extension, res_width=res_width)
        if pdfa is not None:
            pdfa.close()
        text = base64.b64encode(text.read())
        text = text.decode('ascii')
        return {'documentId': doc_id, 'content': str(text), 'errors': errors}
    finally:
        try:
            text.close()
        except:
            gc.collect()
Vielen Dank für eure Hilfe :)
Benutzeravatar
Felix92
User
Beiträge: 133
Registriert: Mittwoch 7. November 2018, 17:57

Closed
Antworten