def create()

in src/dispatch/document/service.py [0:0]

• 62 lines of code
• 8 McCabe index (conditional complexity)

def create(*, db_session, document_in: DocumentCreate) -> Document:
    """Creates a new document."""
    # handle the special case of only allowing 1 FAQ / Forms Export document per-project
    project = project_service.get_by_name_or_raise(
        db_session=db_session, project_in=document_in.project
    )

    if document_in.resource_type == DocumentResourceReferenceTypes.faq:
        faq_doc = (
            db_session.query(Document)
            .filter(Document.resource_type == DocumentResourceReferenceTypes.faq)
            .filter(Document.project_id == project.id)
            .one_or_none()
        )
        if faq_doc:
            raise ValidationError(
                [
                    ErrorWrapper(
                        ExistsError(
                            msg="FAQ document already defined for this project.",
                            document=faq_doc.name,
                        ),
                        loc="document",
                    )
                ],
                model=DocumentCreate,
            )

    if document_in.resource_type == DocumentResourceTemplateTypes.forms:
        forms_doc = (
            db_session.query(Document)
            .filter(Document.resource_type == DocumentResourceTemplateTypes.forms)
            .filter(Document.project_id == project.id)
            .one_or_none()
        )
        if forms_doc:
            raise ValidationError(
                [
                    ErrorWrapper(
                        ExistsError(
                            msg="Forms export template document already defined for this project.",
                            document=forms_doc.name,
                        ),
                        loc="document",
                    )
                ],
                model=DocumentCreate,
            )

    filters = [
        search_filter_service.get(db_session=db_session, search_filter_id=f.id)
        for f in document_in.filters
    ]

    tags = []
    for t in document_in.tags:
        tags.append(tag_service.get_or_create(db_session=db_session, tag_in=t))

    # set the last reminder to now
    if document_in.evergreen:
        document_in.evergreen_last_reminder_at = datetime.utcnow()

    document = Document(
        **document_in.dict(exclude={"project", "filters", "tags"}),
        filters=filters,
        project=project,
        tags=tags,
    )

    db_session.add(document)
    db_session.commit()
    return document