mirror of
https://github.com/docling-project/docling-serve.git
synced 2025-11-29 16:43:24 +00:00
fix: processing complex params in multipart-form (#210)
Signed-off-by: Michele Dolfi <dol@zurich.ibm.com>
This commit is contained in:
@@ -1,9 +1,30 @@
|
||||
import inspect
|
||||
import json
|
||||
import re
|
||||
from typing import Union
|
||||
from typing import Union, get_args, get_origin
|
||||
|
||||
from fastapi import Depends, Form
|
||||
from pydantic import BaseModel
|
||||
from pydantic import BaseModel, TypeAdapter
|
||||
|
||||
|
||||
def is_pydantic_model(type_):
|
||||
try:
|
||||
if inspect.isclass(type_) and issubclass(type_, BaseModel):
|
||||
return True
|
||||
|
||||
origin = get_origin(type_)
|
||||
if origin is Union:
|
||||
args = get_args(type_)
|
||||
return any(
|
||||
inspect.isclass(arg) and issubclass(arg, BaseModel)
|
||||
for arg in args
|
||||
if arg is not type(None)
|
||||
)
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
|
||||
# Adapted from
|
||||
@@ -12,25 +33,62 @@ def FormDepends(cls: type[BaseModel]):
|
||||
new_parameters = []
|
||||
|
||||
for field_name, model_field in cls.model_fields.items():
|
||||
annotation = model_field.annotation
|
||||
description = model_field.description
|
||||
default = (
|
||||
Form(..., description=description)
|
||||
if model_field.is_required()
|
||||
else Form(
|
||||
model_field.default,
|
||||
examples=model_field.examples,
|
||||
description=description,
|
||||
)
|
||||
)
|
||||
|
||||
# Flatten nested Pydantic models by accepting them as JSON strings
|
||||
if is_pydantic_model(annotation):
|
||||
annotation = str
|
||||
default = Form(
|
||||
None
|
||||
if model_field.default is None
|
||||
else json.dumps(model_field.default.model_dump(mode="json")),
|
||||
description=description,
|
||||
examples=None
|
||||
if not model_field.examples
|
||||
else [
|
||||
json.dumps(ex.model_dump(mode="json"))
|
||||
for ex in model_field.examples
|
||||
],
|
||||
)
|
||||
|
||||
new_parameters.append(
|
||||
inspect.Parameter(
|
||||
name=field_name,
|
||||
kind=inspect.Parameter.POSITIONAL_ONLY,
|
||||
default=(
|
||||
Form(...)
|
||||
if model_field.is_required()
|
||||
else Form(model_field.default)
|
||||
),
|
||||
annotation=model_field.annotation,
|
||||
default=default,
|
||||
annotation=annotation,
|
||||
)
|
||||
)
|
||||
|
||||
async def as_form_func(**data):
|
||||
for field_name, model_field in cls.model_fields.items():
|
||||
value = data.get(field_name)
|
||||
annotation = model_field.annotation
|
||||
|
||||
# Parse nested models from JSON string
|
||||
if value is not None and is_pydantic_model(annotation):
|
||||
try:
|
||||
validator = TypeAdapter(annotation)
|
||||
data[field_name] = validator.validate_json(value)
|
||||
except Exception as e:
|
||||
raise ValueError(f"Invalid JSON for field '{field_name}': {e}")
|
||||
|
||||
return cls(**data)
|
||||
|
||||
sig = inspect.signature(as_form_func)
|
||||
sig = sig.replace(parameters=new_parameters)
|
||||
as_form_func.__signature__ = sig # type: ignore
|
||||
|
||||
return Depends(as_form_func)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user