`presentation.routers.api.v1.imports`¶

src.presentation.routers.api.v1.imports ¶

Imports resource handlers.

Handler functions for file-based data imports. Routes are registered via ROUTE_REGISTRY in routes/registry.py.

Handlers

import_from_file - Import data from uploaded file list_supported_formats - List supported file formats

Reference

docs/architecture/api-design-patterns.md
docs/guides/chase-file-import.md

Classes¶

Functions¶

import_from_file `async` ¶

import_from_file(
    request: Request,
    current_user: AuthenticatedUser,
    file: Annotated[
        UploadFile,
        File(description="Financial data file (QFX, OFX)"),
    ],
    handler: ImportFromFileHandler = Depends(
        handler_factory(ImportFromFileHandler)
    ),
) -> ImportResponse | JSONResponse

Import financial data from an uploaded file.

POST /api/v1/imports → 201 Created

Supported formats: - QFX (Quicken Financial Exchange) - Chase Bank - OFX (Open Financial Exchange) - Chase Bank

The file format and provider are auto-detected from the file extension.

Parameters:

Name	Type	Description	Default
`request`	`Request`	FastAPI request object.	required
`current_user`	`AuthenticatedUser`	Authenticated user (from JWT).	required
`file`	`Annotated[UploadFile, File(description='Financial data file (QFX, OFX)')]`	Uploaded file.	required
`handler`	`ImportFromFileHandler`	Import handler (injected).	`Depends(handler_factory(ImportFromFileHandler))`

Returns:

Type	Description
`ImportResponse \| JSONResponse`	ImportResponse with import counts.
`ImportResponse \| JSONResponse`	JSONResponse with RFC 9457 error on failure.

Source code in src/presentation/routers/api/v1/imports.py

async def import_from_file(
    request: Request,
    current_user: AuthenticatedUser,
    file: Annotated[UploadFile, File(description="Financial data file (QFX, OFX)")],
    handler: ImportFromFileHandler = Depends(handler_factory(ImportFromFileHandler)),
) -> ImportResponse | JSONResponse:
    """Import financial data from an uploaded file.

    POST /api/v1/imports → 201 Created

    Supported formats:
    - QFX (Quicken Financial Exchange) - Chase Bank
    - OFX (Open Financial Exchange) - Chase Bank

    The file format and provider are auto-detected from the file extension.

    Args:
        request: FastAPI request object.
        current_user: Authenticated user (from JWT).
        file: Uploaded file.
        handler: Import handler (injected).

    Returns:
        ImportResponse with import counts.
        JSONResponse with RFC 9457 error on failure.
    """
    # Get filename and extension
    filename = file.filename or "unknown"
    extension = ""
    if "." in filename:
        extension = "." + filename.rsplit(".", 1)[-1].lower()

    # Auto-detect format and provider from extension
    if extension not in EXTENSION_MAP:
        # 415 Unsupported Media Type - use JSONResponse directly
        # since ErrorResponseBuilder maps status from error code
        return JSONResponse(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            content={
                "type": f"/errors/{ApplicationErrorCode.COMMAND_VALIDATION_FAILED.value}",
                "title": "Unsupported Media Type",
                "status": status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
                "detail": f"Unsupported file format: {extension}. Supported: {', '.join(EXTENSION_MAP.keys())}",
                "instance": str(request.url.path),
                "trace_id": get_trace_id() or "",
            },
        )

    file_format, provider_slug = EXTENSION_MAP[extension]

    # Read file content
    file_content = await file.read()

    if not file_content:
        return ErrorResponseBuilder.from_application_error(
            error=ApplicationError(
                code=ApplicationErrorCode.COMMAND_VALIDATION_FAILED,
                message="Empty file",
            ),
            request=request,
            trace_id=get_trace_id() or "",
        )

    # Create and execute command
    command = ImportFromFile(
        user_id=current_user.user_id,
        provider_slug=provider_slug,
        file_content=file_content,
        file_format=file_format,
        file_name=filename,
    )

    result = await handler.handle(command)

    if isinstance(result, Failure):
        app_error = _map_import_error(result.error)
        return ErrorResponseBuilder.from_application_error(
            error=app_error,
            request=request,
            trace_id=get_trace_id() or "",
        )

    return ImportResponse.from_result(result.value)

list_supported_formats `async` ¶

list_supported_formats() -> SupportedFormatsResponse

List supported file formats for import.

GET /api/v1/imports/formats → 200 OK