Validation

filestore validates every upload before persisting it. Validation failures are captured per-file in the Store result — they don't crash your endpoint.

File Size Limits

from filestore import FileStore, LocalEngine, StoreConfig

storage = FileStore(
    "document",
    engine=LocalEngine(base_dir="uploads"),
    config=StoreConfig(
        max_file_size=10 * 1024 * 1024,  # 10 MB
        min_file_size=1,                  # At least 1 byte (reject empty files)
    ),
)

Bounds are validated at startup — a config with min_file_size > max_file_size raises immediately.

Streaming validation

For LocalEngine, size limits are checked during the write — not after. If a file exceeds max_file_size mid-stream, the write is aborted immediately and the temp file is cleaned up. Cloud engines verify the real stream size before uploading; the client-reported size hint is only used as an early check.

Extension Allow-List

config = StoreConfig(
    allowed_extensions=[".jpg", ".jpeg", ".png", ".webp"],
)

Extensions are case-insensitive and normalized at construction. The leading dot is optional — "png" and ".PNG" both become ".png".

You can also pass a single string:

config = StoreConfig(allowed_extensions=".pdf")

Content-Type Allow-List

config = StoreConfig(
    allowed_content_types=["image/jpeg", "image/png", "image/webp"],
)

Client-reported types

The content type comes from the client's Content-Type header. It is not verified against the actual file content. For security-critical validation, combine this with a custom filter.

Custom Filters

For validation logic that goes beyond size and type, use filter callbacks. Filters receive a single UploadContext argument:

from filestore import FileStore, MemoryEngine, StoreConfig, UploadContext


async def no_executables(ctx: UploadContext):
    """Reject files with dangerous extensions."""
    dangerous = {".exe", ".bat", ".cmd", ".sh", ".ps1"}
    ext = (ctx.file.filename or "").rsplit(".", 1)[-1].lower()
    if f".{ext}" in dangerous:
        return f"Executable files are not allowed: {ctx.file.filename}"
    return True


storage = FileStore(
    "file",
    engine=MemoryEngine(),
    config=StoreConfig(filters=[no_executables]),
)

Filter Return Values

Return Value	Effect
True	Accept the file, continue to next filter
False	Reject the file with a generic message
"Custom message"	Reject the file with the given message

Multiple Filters

Filters run in order. The first rejection stops the chain:

config = StoreConfig(
    filters=[
        check_file_magic,      # Run first
        check_virus_scan,      # Run second (only if first passed)
        check_content_policy,  # Run third
    ],
)

Store-level and field-level filters are concatenated — store filters run first.

Sync and Async

Filters can be sync or async — filestore handles both:

# Sync filter
def check_size(ctx):
    return True

# Async filter
async def check_virus(ctx):
    result = await virus_scanner.scan(ctx.file)
    return result.is_clean or "File failed virus scan"

Combining Validation

All validation types work together:

config = StoreConfig(
    destination="uploads/images",
    allowed_extensions=[".jpg", ".png"],
    allowed_content_types=["image/jpeg", "image/png"],
    max_file_size=5 * 1024 * 1024,
    min_file_size=100,
    filters=[custom_image_validator],
)

Validation runs in this order:

1. Extension check — based on the resolved (final) filename
2. Content-type check — based on the client-reported MIME type
3. Size check (pre-upload) — based on the client-reported size hint
4. Custom filters — your callbacks
5. Size check (during/before persistence) — verified against actual bytes

Handling Validation Failures

Failed files don't raise exceptions. They appear in the Store result:

@app.post("/upload")
async def upload(store: Store = Depends(storage)):
    if not store:  # anything but UploadStatus.COMPLETED
        return {"status": store.status, "errors": store.errors}

    for file_data in store.failed_files:
        print(f"Rejected: {file_data.original_filename} — {file_data.error}")

    for file_data in store.successful_files:
        print(f"Saved: {file_data.filename} ({file_data.size} bytes)")