File Storage

AMSDAL provides a File model and pluggable storage backends for managing file uploads and downloads.

File Model

File is a regular AMSDAL Model — it supports all standard operations like save(), asave(), objects.filter(), etc. When a File is saved, AMSDAL automatically calls pre_create()/pre_update() hooks that persist the file content to the configured storage backend before writing the record to the database.

Creating files

from amsdal.models.core.file import File

# From bytes
file = File.from_bytes('document.txt', b'Hello, World!')
file.save()

# From a file path
from pathlib import Path
file = File.from_file(Path('/path/to/report.pdf'))
file.save()

# From a file object
with open('image.png', 'rb') as f:
    file = File.from_file(f)
    file.save()

Reading files

# Read all bytes
content = file.read_bytes()

# Open as stream
with file.open() as f:
    content = f.read()

# Write to disk
file.to_file(Path('/tmp/output.pdf'))

# Get MIME type
print(file.mimetype)  # e.g. 'application/pdf'

Getting URLs

url = file.url()

The URL format depends on the storage backend:

• FileSystemStorage: https://cdn.example.com/media/document.txt
• DBStorage: db://document.txt (not publicly accessible)
• S3Storage: presigned S3 URL

Async operations

# Save
file = File.from_bytes('doc.txt', b'content')
await file.asave()

# Read
content = await file.aread_bytes()

# Open
async with await file.aopen() as f:
    content = await f.read()

# URL
url = await file.aurl()

---

Storage Backends

DBStorage

Stores file bytes in the database alongside the File record's data field. This is the default backend.

from amsdal_models.storage.backends.db import DBStorage

class MyModel(Model):
    attachment: File = FileField(storage=DBStorage())

• No external storage infrastructure needed
• Files stored in the same database as your models
• Not suitable for large files or high-traffic scenarios

FileSystemStorage

Stores files on the local filesystem.

from amsdal.storages.file_system import FileSystemStorage

class MyModel(Model):
    attachment: File = FileField(
        storage=FileSystemStorage('/var/media', 'https://cdn.example.com/media/'),
    )

Parameters:

Parameter	Description
base_dir	Root directory for stored files. Falls back to MEDIA_ROOT setting.
base_url	URL prefix for public access. Falls back to MEDIA_URL setting.

If base_dir and base_url are omitted, the values from settings are used:

# Uses MEDIA_ROOT and MEDIA_URL from settings
storage = FileSystemStorage()

S3Storage (via amsdal_storages)

Stores files in Amazon S3. Install the amsdal_storages package:

pip install amsdal_storages

from amsdal_storages.s3.storage import S3Storage

class MyModel(Model):
    attachment: File = FileField(
        storage=S3Storage(
            bucket='my-bucket',
            region_name='us-east-1',
            object_prefix='uploads/',
            presign_ttl=3600,
        ),
    )

---

Settings

Configure default file storage behavior via settings or environment variables:

Setting	Env Variable	Default	Description
MEDIA_ROOT	AMSDAL_MEDIA_ROOT	./media	Root directory for filesystem storage
MEDIA_URL	AMSDAL_MEDIA_URL	/media/	URL prefix for file access
DEFAULT_FILE_STORAGE	AMSDAL_DEFAULT_FILE_STORAGE	amsdal_models.storage.backends.db.DBStorage	Default storage backend class

---

Custom Storage Backend

Create a custom backend by subclassing Storage:

from amsdal_models.storage.base import Storage
from amsdal_models.storage.types import FileProtocol
from typing import BinaryIO, IO, Any

class MyCloudStorage(Storage):
    keeps_local_copy = False

    def __init__(self, bucket: str, **kwargs):
        self.bucket = bucket

    def save(self, file: FileProtocol, content: BinaryIO) -> str:
        # Upload content to your cloud provider
        key = f'{self.bucket}/{file.filename}'
        self._upload(key, content)
        return file.filename

    def open(self, file: FileProtocol, mode: str = 'rb') -> IO[Any]:
        # Return a file-like stream from your cloud provider
        return self._download_stream(file.filename)

    def delete(self, file: FileProtocol) -> None:
        self._remove(file.filename)

    def exists(self, file: FileProtocol) -> bool:
        return self._check_exists(file.filename)

    def url(self, file: FileProtocol) -> str:
        return f'https://{self.bucket}.cloud.example.com/{file.filename}'

    def _export_kwargs(self) -> dict[str, Any]:
        return {'bucket': self.bucket}

For async support, also implement asave(), aopen(), adelete(), aexists(), and aurl().

Setting as default

from amsdal.storages import set_default_storage

set_default_storage(MyCloudStorage(bucket='my-app'))

Or configure via environment:

AMSDAL_DEFAULT_FILE_STORAGE=myapp.storages.MyCloudStorage