def create_storage()

in src/dispatch/storage/flows.py [0:0]

• 64 lines of code
• 19 McCabe index (conditional complexity)

def create_storage(subject: Subject, storage_members: List[str], db_session: Session):
    """Creates a storage."""
    plugin = plugin_service.get_active_instance(
        db_session=db_session, project_id=subject.project.id, plugin_type="storage"
    )
    if not plugin:
        log.warning("Storage not created. No storage plugin enabled.")
        return

    external_storage_root_id = None

    # if project is set to use the tag external_id for the root folder
    storage_tag_type = tag_type_service.get_storage_tag_type_for_project(
        db_session=db_session,
        project_id=subject.project.id,
    )

    if storage_tag_type:
        # find if the subject has a tag of the specified type
        tag = next((tag for tag in subject.tags if tag.tag_type.id == storage_tag_type.id), None)
        if tag:
            external_storage_root_id = tag.external_id

    if external_storage_root_id is None:
        # we create the external storage based on the root in the configuration of the plugin
        external_storage_root_id = plugin.configuration.root_id

    storage_name = subject.title if subject.project.storage_use_title else subject.name

    try:
        external_storage = plugin.instance.create_file(
            parent_id=external_storage_root_id, name=storage_name, participants=storage_members
        )
    except Exception as e:
        log.exception(e)
        return

    if not external_storage:
        log.error(f"Storage not created. Plugin {plugin.plugin.slug} encountered an error.")
        return

    # we create folders to store logs and screengrabs
    folder_one_name = (
        subject.project.storage_folder_one if subject.project.storage_folder_one else "Logs"
    )
    folder_one = plugin.instance.create_file(parent_id=external_storage["id"], name=folder_one_name)

    folder_two_name = (
        subject.project.storage_folder_two if subject.project.storage_folder_two else "Screengrabs"
    )
    plugin.instance.create_file(parent_id=external_storage["id"], name=folder_two_name)

    if subject.project.storage_use_folder_one_as_primary:
        external_storage = folder_one

    # we create the internal storage
    storage_in = StorageCreate(
        resource_id=external_storage["id"],
        resource_type=plugin.plugin.slug,
        weblink=external_storage["weblink"],
    )

    storage = create(db_session=db_session, storage_in=storage_in)
    subject.storage = storage
    db_session.add(subject)
    db_session.commit()

    subject_type = get_table_name_by_class_instance(subject)
    if subject_type == "case":
        event_service.log_case_event(
            db_session=db_session,
            source=plugin.plugin.title,
            description="Case storage created",
            case_id=subject.id,
        )
    if subject_type == "incident":
        event_service.log_incident_event(
            db_session=db_session,
            source=plugin.plugin.title,
            description="Incident storage created",
            incident_id=subject.id,
        )

    return storage