File Storage

The openviper.storage package provides a pluggable file storage API for handling uploaded files. The default FileSystemStorage persists files under MEDIA_ROOT and builds public URLs from MEDIA_URL.

Overview

All storage operations are coroutine-based and run file I/O in thread pools via aiofiles. The default_storage singleton is configured from settings and used automatically by FileField and ImageField.

Key Classes

class openviper.storage.Storage

Protocol defining the interface for all storage backends.

CHUNK_SIZE

Size of chunks for streaming I/O (class-level int).

save(name, content) → Awaitable[str]

Persist content (bytes, file-like, or async iterator) under name. Returns the final relative path (may differ from name to avoid collisions).

delete(name) → Awaitable[None]

Delete the file at name.

exists(name) → Awaitable[bool]

Return True if name exists in the storage.

url(name) → str

Return the public URL for name.

size(name) → Awaitable[int]

Return the size in bytes of name.

read(name) → Awaitable[bytes]

Return the full file content. Raises ValueError when the file exceeds MAX_READ_SIZE.

listdir(path='') → Awaitable[list[str]]

List entries under path in storage.

class openviper.storage.FileSystemStorage(location=None, base_url=None, chunk_size=1048576)

Concrete storage that persists files to the local filesystem.

• location - directory for files (defaults to settings.MEDIA_ROOT).

• base_url - URL prefix for generated URLs (defaults to settings.MEDIA_URL).

• chunk_size - chunk size for streaming uploads (default: 1 MiB).

File names are sanitised to remove unsafe characters, and duplicate names are disambiguated by appending a UUID suffix.

save(name, content) → Awaitable[str]

Persist content using an atomic write pattern (temp file + rename). Re-verifies path containment after temp file creation to mitigate TOCTOU.

delete(name) → Awaitable[None]

Delete the file at name. No error if it does not exist.

exists(name) → Awaitable[bool]

Return True if name exists in the storage.

url(name) → str

Return the public URL with percent-encoded path segments.

size(name) → Awaitable[int]

Return the size in bytes. Raises FileNotFoundError if the file does not exist.

read(name) → Awaitable[bytes]

Return the full file content. Raises ValueError when the file exceeds MAX_READ_SIZE (100 MiB).

listdir(path='') → Awaitable[list[str]]

List entries under path in storage.

Utilities

openviper.storage.generate_unique_name(name) → str

Generate a collision-resistant file name by appending a full UUID hex suffix. Used internally by FileSystemStorage.save() when the target name already exists.

Module Constants

Name	Description
UNSAFE_FILENAME_RE	Compiled regex matching characters disallowed in file names.
HIDDEN_FILENAME_RE	Compiled regex matching leading-dot (hidden) file names.
MAX_COMPONENT_LEN	Maximum length of a single path component (255).
MAX_READ_SIZE	Maximum file size for FileSystemStorage.read() (100 MiB).

Type Aliases

Name	Type
StorageContent	bytes | bytearray | AsyncIterator[bytes] | IO

Default Storage Proxy

class openviper.storage.DefaultStorage

Thread-safe lazy proxy that creates a FileSystemStorage on first access.

configure(storage) → None

Programmatically override the default storage backend.

Delegates all Storage methods (save, delete, exists, url, size, read, listdir) to the underlying instance.

openviper.storage.default_storage

Module-level DefaultStorage singleton.

Example Usage

Using default_storage Directly

from openviper.storage import default_storage
from openviper.http.request import Request
from openviper.http.response import JSONResponse
from openviper.routing.router import Router

router = Router()

@router.post("/upload")
async def upload_file(request: Request) -> JSONResponse:
    files = await request.files()
    file = files.get("file")
    if file is None:
        return JSONResponse({"error": "No file provided"}, status_code=400)

    content = await file.read()
    saved_path = await default_storage.save(f"uploads/{file.filename}", content)
    url = default_storage.url(saved_path)

    return JSONResponse({"path": saved_path, "url": url})

FileField on a Model

When a model has a FileField, the framework automatically uses default_storage to persist the file on save():

from openviper.db.models import Model
from openviper.db import fields

class Document(Model):
    class Meta:
        table_name = "documents"

    title = fields.CharField(max_length=255)
    file = fields.FileField(upload_to="documents/")

# In a view:
@router.post("/documents")
async def upload_doc(request: Request) -> JSONResponse:
    form = await request.form()
    uploaded = (await request.files()).get("file")
    doc = Document(title=form["title"], file=uploaded)
    await doc.save()
    return JSONResponse({"url": default_storage.url(doc.file)}, status_code=201)

Custom Storage Backend

Implement the Storage protocol to integrate any cloud or third-party storage:

from openviper.storage.base import Storage

class S3Storage:
    def __init__(self, bucket: str) -> None:
        self.bucket = bucket

    async def save(self, name: str, content: bytes) -> str:
        # Upload to S3 ...
        return name

    async def delete(self, name: str) -> None:
        # Delete from S3 ...
        pass

    def url(self, name: str) -> str:
        return f"https://{self.bucket}.s3.amazonaws.com/{name}"

    async def exists(self, name: str) -> bool:
        # Check S3 ...
        return False

    async def size(self, name: str) -> int:
        return 0

    async def read(self, name: str) -> bytes:
        # Fetch from S3 ...
        return b""

    async def listdir(self, path: str = "") -> list[str]:
        return []

Configuration

@dataclasses.dataclass(frozen=True)
class MySettings(Settings):
    MEDIA_ROOT: str = "media"
    MEDIA_URL: str = "/media/"
    DEFAULT_FILE_STORAGE: str = "myproject.storage.S3Storage"

API Reference

OpenViper file storage backends.

class openviper.storage.FileSystemStorage(location=None, base_url=None, chunk_size=1048576)

Bases: object

Store files on the local filesystem with async I/O and streaming.

Parameters:

• location (str | None) – Directory path for file storage. Defaults to MEDIA_ROOT from settings.

• base_url (str | None) – URL prefix for serving files. Defaults to MEDIA_URL from settings.

• chunk_size (int) – Chunk size for streaming uploads (default: 1 MiB).

CHUNK_SIZE: int = 1048576

property location: str

property base_url: str

validate_name(name)

Validate and sanitise name against path traversal.

• Rejects null bytes.

• Replaces hidden filenames (leading dot).

• Normalises separators.

• Removes .. components (path traversal guard).

• Truncates each component to MAX_COMPONENT_LEN.

Raises:

ValueError – if name is empty, contains null bytes, or resolves to an empty path after sanitisation.

Parameters:

name (str)

Return type:

str

full_path(name)

Return absolute path, guaranteed to be inside location.

Performs symlink detection and path containment verification to prevent directory traversal attacks. Containment is checked before symlink inspection so that escape attempts are rejected regardless of symlink state.

Raises:

ValueError – if the resolved path escapes the storage root, or if the path involves symlinks.

Parameters:

name (str)

Return type:

Path

async mkdir_async(path)

Create path directory (and parents) asynchronously.

Parameters:

path (Path)

Return type:

None

resolved_path(name)

Validate name and return its absolute path inside location.

Combines validate_name() and full_path() into a single call so callers do not repeat the two-step dance.

Parameters:

name (str)

Return type:

Path

async save(name, content)

Save file with async I/O and chunked uploads.

Uses an atomic write pattern (temp file + rename) to prevent partial writes. Re-verifies path containment after temp file creation to mitigate TOCTOU.

Parameters:

• name (str)

• content (StorageContent)

Return type:

str

async delete(name)

Delete file asynchronously. No error if it does not exist.

Parameters:

name (str)

Return type:

None

async exists(name)

Check if file exists asynchronously.

Parameters:

name (str)

Return type:

bool

url(name)

Return the public URL with percent-encoded path segments.

Parameters:

name (str)

Return type:

str

async size(name)

Get file size asynchronously.

Parameters:

name (str)

Return type:

int

async read(name)

Read and return the full content of the file at name.

Raises:

ValueError – if the file exceeds MAX_READ_SIZE bytes.

Parameters:

name (str)

Return type:

bytes

async listdir(path='')

List entries under path in storage.

Parameters:

path (str)

Return type:

list[str]

class openviper.storage.Storage(*args, **kwargs)

Bases: Protocol

Protocol defining the interface for all storage backends.

CHUNK_SIZE: int

async save(name, content)

Parameters:

• name (str)

• content (StorageContent)

Return type:

str

async delete(name)

Parameters:

name (str)

Return type:

None

async exists(name)

Parameters:

name (str)

Return type:

bool

url(name)

Parameters:

name (str)

Return type:

str

async size(name)

Parameters:

name (str)

Return type:

int

async read(name)

Parameters:

name (str)

Return type:

bytes

async listdir(path='')

Parameters:

path (str)

Return type:

list[str]

openviper.storage.generate_unique_name(name)

Generate a collision-resistant name via UUID hex suffix.

Parameters:

name (str)

Return type:

str

File storage backends for OpenViper.

Pluggable storage API for handling uploaded files. FileSystemStorage persists to MEDIA_ROOT and serves at MEDIA_URL.

type openviper.storage.base.StorageContent = bytes | bytearray | AsyncIterator[bytes] | IO

class openviper.storage.base.Storage(*args, **kwargs)

Bases: Protocol

Protocol defining the interface for all storage backends.

CHUNK_SIZE: int

async save(name, content)

Parameters:

• name (str)

• content (StorageContent)

Return type:

str

async delete(name)

Parameters:

name (str)

Return type:

None

async exists(name)

Parameters:

name (str)

Return type:

bool

url(name)

Parameters:

name (str)

Return type:

str

async size(name)

Parameters:

name (str)

Return type:

int

async read(name)

Parameters:

name (str)

Return type:

bytes

async listdir(path='')

Parameters:

path (str)

Return type:

list[str]

openviper.storage.base.generate_unique_name(name)

Generate a collision-resistant name via UUID hex suffix.

Parameters:

name (str)

Return type:

str

class openviper.storage.base.FileSystemStorage(location=None, base_url=None, chunk_size=1048576)

Bases: object

Store files on the local filesystem with async I/O and streaming.

Parameters:

• location (str | None) – Directory path for file storage. Defaults to MEDIA_ROOT from settings.

• base_url (str | None) – URL prefix for serving files. Defaults to MEDIA_URL from settings.

• chunk_size (int) – Chunk size for streaming uploads (default: 1 MiB).

CHUNK_SIZE: int = 1048576

property location: str

property base_url: str

validate_name(name)

Validate and sanitise name against path traversal.

• Rejects null bytes.

• Replaces hidden filenames (leading dot).

• Normalises separators.

• Removes .. components (path traversal guard).

• Truncates each component to MAX_COMPONENT_LEN.

Raises:

ValueError – if name is empty, contains null bytes, or resolves to an empty path after sanitisation.

Parameters:

name (str)

Return type:

str

full_path(name)

Return absolute path, guaranteed to be inside location.

Performs symlink detection and path containment verification to prevent directory traversal attacks. Containment is checked before symlink inspection so that escape attempts are rejected regardless of symlink state.

Raises:

ValueError – if the resolved path escapes the storage root, or if the path involves symlinks.

Parameters:

name (str)

Return type:

Path

async mkdir_async(path)

Create path directory (and parents) asynchronously.

Parameters:

path (Path)

Return type:

None

resolved_path(name)

Validate name and return its absolute path inside location.

Combines validate_name() and full_path() into a single call so callers do not repeat the two-step dance.

Parameters:

name (str)

Return type:

Path

async save(name, content)

Save file with async I/O and chunked uploads.

Uses an atomic write pattern (temp file + rename) to prevent partial writes. Re-verifies path containment after temp file creation to mitigate TOCTOU.

Parameters:

• name (str)

• content (StorageContent)

Return type:

str

async delete(name)

Delete file asynchronously. No error if it does not exist.

Parameters:

name (str)

Return type:

None

async exists(name)

Check if file exists asynchronously.

Parameters:

name (str)

Return type:

bool

url(name)

Return the public URL with percent-encoded path segments.

Parameters:

name (str)

Return type:

str

async size(name)

Get file size asynchronously.

Parameters:

name (str)

Return type:

int

async read(name)

Read and return the full content of the file at name.

Raises:

ValueError – if the file exceeds MAX_READ_SIZE bytes.

Parameters:

name (str)

Return type:

bytes

async listdir(path='')

List entries under path in storage.

Parameters:

path (str)

Return type:

list[str]

class openviper.storage.base.DefaultStorage

Bases: object

Thread-safe lazy proxy for the default storage backend.

get_storage()

Return type:

FileSystemStorage

configure(storage)

Programmatically override the default storage backend.

Parameters:

storage (FileSystemStorage)

Return type:

None

async save(name, content)

Parameters:

• name (str)

• content (StorageContent)

Return type:

str

async delete(name)

Parameters:

name (str)

Return type:

None

async exists(name)

Parameters:

name (str)

Return type:

bool

url(name)

Parameters:

name (str)

Return type:

str

async size(name)

Parameters:

name (str)

Return type:

int

async read(name)

Parameters:

name (str)

Return type:

bytes

async listdir(path='')

Parameters:

path (str)

Return type:

list[str]