orders

update_documents_data async

update_documents_data(
    order_id: ObjectId,
    order: ConfirmedOrderUpdateModel
    | OrderCreateModel
    | OrderUpdateModel,
    order_data: dict,
    carrier_container: dict | bool = False,
)

:param carrier_container: is used to get existing carrier from order data False if it is not possible, True if needs to be retrieved, order_data if it is queried already

Source code in api/operations/orders.py

async def update_documents_data(
    order_id: ObjectId,
    order: ConfirmedOrderUpdateModel | OrderCreateModel | OrderUpdateModel,
    order_data: dict,
    carrier_container: dict | bool = False,
):
    """
    :param carrier_container:  is used to get existing carrier from order data
        False if it is not possible,
        True if needs to be retrieved,
        order_data if it is queried already
    """
    if not order.documents:
        return
    order_data["documents"] = []
    doc_insert_ids = [
        ObjectId(document.id)
        for document in order.documents
        if isinstance(document, InsertByOIDModel) and document.id
    ]
    if doc_insert_ids:
        doc_inserts = await docs_col.find_batch(
            {"_id": {"$in": doc_insert_ids}}
        )
        for doc in doc_inserts:
            check_verification(
                doc, f'Document {doc.get("name", "(nameless)")}'
            )
        order_data["documents"].extend(doc_inserts)
    carrier = None
    if carrier_container is True:
        carrier_container = await orders_col.find_single("id", order_id)
    if carrier_container:
        carrier = carrier_container.get("carrier")  # type: ignore[union-attr]
    for doc in order.documents:
        if not isinstance(doc, DocumentCreateModel):
            continue
        doc_data = doc.model_dump()
        doc_data["created_at"] = get_current_datetime()
        doc_data["object_id"] = order_id
        doc_data["collection"] = "orders"
        # fixme organization_id of person who loads
        doc_data[DocumentDBModel.text_search] = get_documents_text_search(
            DocumentDBModel(**doc_data)
        )
        if carrier:
            doc_data["organization_id"] = carrier.get("id")
        doc_created = await docs_col.create(
            DocumentDBModel(**doc_data).model_dump()
        )
        order_data["documents"].append(doc_created)