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}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()