Storage

Storage (sillo.storage)

One way to put files somewhere and get them back, whether “somewhere” is a directory on this machine or a bucket in object storage. Moving between the two is a configuration change, not a code change.

from sillo.storage import BucketConfig, StorageConfig, setup_storage
from sillo.storage.policies import Owned, Private

storage = setup_storage(app, StorageConfig(
    default="attachments",
    buckets={
        "attachments": BucketConfig(
            driver="local",
            root="storage/attachments",
            policy=Private(),
        ),
        "avatars": BucketConfig(
            driver="local",
            root="storage/avatars",
            policy=Owned(),
            accepts=("image/png", "image/jpeg"),
            max_bytes=2 * 1024 * 1024,
        ),
    },
))

Then, in a handler:

from sillo.storage import bucket, stream_upload

async def upload_avatar(request, response):
    upload = (await request.files)["avatar"]

    stored = await bucket("avatars").put(
        f"{request.user.identity}/face.png",
        stream_upload(upload),
        content_type=upload.content_type or "",
        user=request.user,
    )

    return response.json({"key": stored.key, "size": stored.size})

bucket(name) just works — in this handler, in a queue job, in a script — with nothing to set up beyond calling setup_storage once at startup. There is no object to fetch first and no request to have in hand; import it and call it. (If you’re curious how, that’s in Instance Registry — nothing here depends on knowing.)

The only reason to hold onto what setup_storage returns is for the handful of things bucket() doesn’t cover — .listen() below, or building a second bucket at runtime:

storage = setup_storage(app, StorageConfig(...))

What ships

Driver	Dependency	Use
local	none	Files under a directory. What you start with.
memory	none	A dictionary. Tests, and small ephemera.
s3	uv add "sillo-framework[storage-s3]"	Anything speaking the S3 API.

s3 covers AWS S3, MinIO, Cloudflare R2, Backblaze B2, DigitalOcean Spaces, Wasabi and Ceph — every one of those is a different endpoint and nothing else in your code.

Azure and native GCS

Neither has an S3 API that is complete enough to rely on. Both need a driver of their own, which is what the driver contract is for — about two hundred lines against a test suite that will tell you when you are finished.

The five properties

Everything in this package follows from five decisions. They are worth reading once, because each one shows up in a signature you will use.

One driver contract. Local disk and object storage differ, and the contract is the intersection where the differences cannot leak. That is why page() exists and list() does not.

Streams, never buffers. put takes an async iterator. There is no convenience that takes bytes, because the moment one exists everybody uses it.

Signed URLs are scoped four ways — method, expiry, content type and size. Each one left out is a permission you did not mean to grant.

The declared content type is not evidence. What a file is gets decided by looking at it. What the uploader claimed is kept only so the two can be compared.

Access is a policy object, not a public flag. Two values cannot express “this user, under their own prefix”, which is the rule most applications actually want.

---

Configuring

StorageConfig

Field	Default	Meaning
buckets	{}	Bucket names to their configuration.
default	""	Which bucket storage.bucket() returns unasked.
secret	""	Signs URLs. Falls back to the application’s own secret.
route	"/storage"	Where the serving route is mounted.
serve	True	Mount the serving route at all.

BucketConfig

Field	Default	Meaning
driver	"local"	local, memory or s3.
policy	Private()	Who may do what.
root	""	The directory, for local. Required.
bucket	""	The remote bucket name, for s3.
endpoint	""	The S3 endpoint, for anything that is not AWS.
region	"us-east-1"	The S3 region.
access_key, secret_key	""	S3 credentials. Read these from the environment.
max_bytes	0	Largest object accepted. 0 means no limit.
accepts	()	Content types accepted after sniffing. Empty means anything.

A bucket that names a driver it cannot build fails at setup_storage, not at the first upload:

# ValueError: bucket 'attachments' uses the local driver and has no root
setup_storage(app, StorageConfig(buckets={"attachments": BucketConfig(driver="local")}))

---

Putting things in

put consumes an async iterator once. It never holds the whole file.

stored = await bucket.put("reports/q3.pdf", stream, content_type="application/pdf")

stored.key           # "reports/q3.pdf"
stored.size          # 2_411_984
stored.content_type  # "application/pdf" — what the *sniffer* decided
stored.etag          # the backend's version marker

From an upload

sillo’s multipart parser already spools the body to disk, so stream_upload reads it back in 64 kB pieces and an upload of any size crosses into a bucket without ever being assembled in memory:

from sillo.storage import stream_upload

files = await request.files
upload = files["document"]

await bucket.put(
    f"documents/{upload.filename}",
    stream_upload(upload),
    content_type=upload.content_type or "",
    user=request.user,
)

stream_upload rewinds the upload before reading it. A handler that has already inspected it — checked a magic number, measured it — leaves the cursor somewhere in the middle, and storing from there writes a truncated file with no error anywhere.

Not `await upload.read()`

Reading the whole upload and passing the bytes works, and undoes the reason put takes a stream at all. A 2 GB upload becomes 2 GB of process memory.

From bytes you already have

Deliberately explicit, because it is a decision to buffer:

from sillo.storage import chunks

await bucket.put("notes/hello.txt", chunks(b"hello"), content_type="text/plain")

From another source

Anything that yields bytes works — a generator, an HTTP response, another bucket:

async def rows():
    async for record in Document.all():
        yield f"{record.id},{record.title}\n".encode()

await bucket.put("exports/documents.csv", rows(), content_type="text/csv")

Size limits are enforced mid-stream

max_bytes is checked as bytes arrive, not against a declared Content-Length — which is a number the uploader chose. A body claiming one megabyte and sending a hundred is refused at the megabyte, having buffered none of the rest.

---

Getting things out

get returns an async iterator. There is no await before the async for:

async for chunk in bucket.get("reports/q3.pdf", user=request.user):
    ...

To hand it to a client, stream it:

info = await bucket.stat(key, user=request.user)

return response.stream(
    bucket.get(key, user=request.user),
    content_type=info.content_type,
    headers={"content-length": str(info.size)},
)

To read it into memory, say so — and say how much you are willing to read:

from sillo.storage import collect

data = await collect(bucket.get("config/small.json"), limit=64 * 1024)

Describing, checking, removing

info = await bucket.stat(key, user=user)
info.size            # 2_411_984
info.content_type    # what it is served as
info.declared_type   # what the uploader claimed
info.mistyped        # True when those two disagree
info.modified        # unix timestamp
info.etag

await bucket.exists(key, user=user)     # True / False
await bucket.delete(key, user=user)     # True, or False if it was not there

Deleting something absent is not an error. The caller wanted it gone; it is gone.

---

Listing

Listing is paginated. There is no method that returns everything.

cursor = ""
while True:
    page = await bucket.page("reports/", cursor=cursor, limit=100, user=user)

    for info in page.files:
        print(info.key, info.size)

    if not page.more:
        break
    cursor = page.cursor

page.prefixes holds the pseudo-directories directly beneath the prefix, so a file browser can show folders:

page = await bucket.page("", user=user)
page.prefixes   # ("avatars/", "exports/", "reports/")

Why there is no list()

S3 pages at a thousand keys and a filesystem does not. A method returning a plain list would work in development and fall over the first time a bucket grew past a thousand objects — in production, on somebody’s largest tenant. Putting the cursor in the signature makes that impossible to forget.

The memory driver pages too, for the same reason: a test double more permissive than production hides exactly the bug it exists to catch.

---

Policies

A policy is asked, per operation, with the user in scope. Five ship:

Policy	Read	Write	Signed URLs
Private()	nobody	nobody	any
Public()	anyone	nobody	any
ReadOnly()	signed-in users	nobody	reads
Signed()	nobody	nobody	reads
Owned()	own prefix	own prefix	any

Private() is the default, and is the default on purpose: a bucket that is readable because nobody configured it is the failure this module exists to prevent.

Owned — the one that pays for the design

BucketConfig(driver="local", root="storage/avatars", policy=Owned())

Every user gets a private area of one shared bucket, keyed on their identity:

await bucket.put("114/face.png", stream, user=user_114)   # fine
await bucket.put("999/face.png", stream, user=user_114)   # PolicyRefused

The prefix is a template, so it can be anything with an {id} in it:

Owned(prefix="users/{id}/uploads/")
Owned(readable=True)   # users may read each other's, but still not write

A prefix without {id} is refused at construction — it would give every user the same area and quietly make the bucket shared.

policy.area(user) gives you the prefix to write under:

key = f"{bucket.policy.area(request.user)}face.png"

Writing your own

A policy is any object with two methods, so a closure over your domain model is a policy:

class TeamOwned:
    """Members of a workspace share its folder."""

    def allows(self, action, key, user=None):
        if user is None or not user.is_authenticated:
            return False
        return key.startswith(f"{user.workspace_id}/")

    def signable(self, action):
        return True

allows answers for a user. signable answers for a signed URL, where there is no user — the permission was decided when the URL was minted.

---

Signed URLs

A signed URL grants one narrow permission for a while. Four things are signed, and all four are enforced:

url = bucket.signed_url(
    "reports/q3.pdf",
    method="GET",           # this method and no other
    expires_in=300,         # and only for five minutes
)

For handing somebody an upload slot:

url = bucket.signed_url(
    f"{user.identity}/face.png",
    method="PUT",
    expires_in=600,
    content_type="image/png",   # and only this type
    max_bytes=2 * 1024 * 1024,  # and only this large
)

Each dimension omitted is a permission granted

A signature over the key alone is a URL that reads and overwrites. One without the content type lets somebody handed a slot for image/png store HTML in it. One without a size limit lets them store a hundred gigabytes.

max_bytes falls back to the bucket’s own limit, so a signed upload slot is never wider than the bucket itself.

How it works

The token is opaque and carries its own claims, so verifying it needs no lookup and no shared state. It is bound to the bucket as well as the object: a token minted for avatars will not open anything in exports, even under the same application secret.

Every refusal — expired, tampered, wrong object, wrong method, malformed — returns the same message. Telling an unauthenticated caller which check failed tells them how the signing works.

The signing secret comes from StorageConfig.secret, or from the application’s own secret if that is unset. A secret shorter than sixteen bytes is refused at startup rather than producing forgeable tokens quietly.

---

Content types

The type an uploader declares is a string the uploader chose. What the file is gets decided by looking at the first four kilobytes:

stored = await bucket.put("a.png", stream, content_type="image/png")
stored.content_type   # "text/html" — because the bytes were markup

The claim is kept so the two can be compared:

info = await bucket.stat("a.png")
info.declared_type   # "image/png"
info.content_type    # "text/html"
info.mistyped        # True

mistyped is worth logging. A .png whose bytes are HTML is not usually a mistake — it is the shape of a stored cross-site scripting attempt.

To refuse it outright, state what the bucket accepts. The list is matched against the sniffed type, so declaring image/png and uploading markup is refused rather than stored:

BucketConfig(..., accepts=("image/png", "image/jpeg"))

# StorageError: avatars does not accept text/html; it accepts image/png, image/jpeg

Anything unrecognised becomes application/octet-stream, which browsers download rather than render. Unknown falling back to “download it” is the safe direction.

---

Serving

setup_storage mounts one route at StorageConfig.route. It is what local signed URLs point at, and it is the half of sniffing that makes sniffing a defence rather than a label.

Every response carries:

Header	Why
X-Content-Type-Options: nosniff	Without it the browser re-sniffs and reaches its own conclusion, and the whole chain was pointless.
Content-Disposition	attachment for anything not on a short render-safe list, so an unexpected type downloads instead of executing.
Content-Security-Policy: sandbox	A last line under the other two.
Cross-Origin-Resource-Policy	Not readable from another origin.

Only these render inline: PNG, JPEG, GIF, WebP, PDF, plain text and CSV. Everything else downloads.

A refusal is a 404, not a 403. A 403 on a private bucket confirms the object exists, which is half of what somebody probing for it wants to know.

Put public buckets on another origin

Headers reduce the risk of serving user-supplied files from your own origin. They do not remove it. If a bucket holds files from users and is served publicly, give it a different hostname.

To turn the route off — because S3 serves its own signed URLs, or because you want to serve files yourself:

StorageConfig(serve=False, ...)

---

Watching

Every driver takes listeners, and every completed operation is reported:

@storage.listen
def audit(event):
    log.info("%s %s %s in %.1fms", event.action, event.bucket, event.key, event.duration_ms)

event.bucket       # "avatars"
event.key          # "114/face.png"
event.action       # Action.WRITE
event.driver       # "local"
event.size         # bytes moved
event.duration_ms
event.outcome      # "ok" | "missing" | "error"
event.error

A listener that raises is skipped for that event and nothing else. An observer must not be able to fail a write.

This is what the Foreman dashboard’s Storage panel reads. It exists in version one deliberately: a hook designed in is a hook nobody has to monkeypatch a private method to get.

---

Errors

Exception	When
FileNotFound	No object under that key. Carries .key.
UnsafeKey	A key that would escape the bucket, or that no backend can hold.
PolicyRefused	The bucket’s policy declined. Carries .action and .key.
SignatureInvalid	A signed URL was missing, expired, tampered with, or for something else.
StorageError	The base. Also raised for size and type refusals.

All five inherit from StorageError, so one except catches the subsystem.

---

Keys

Every key is normalised before it reaches a driver, so a//b, ./a/b and a/./b are one object on every backend rather than three on some of them.

Refused outright: empty keys, absolute keys, anything climbing above the bucket, control characters, backslashes, keys over 1024 bytes and segments over 255. Unicode is normalised to NFC, so a macOS upload and a Linux upload of café.pdf are the same object rather than two that look identical in every listing.

from sillo.storage import normalise

normalise("./a//b.txt")     # "a/b.txt"
normalise("../etc/passwd")  # UnsafeKey

Containment is checked by resolving the path and comparing it against the bucket’s root — not by looking for .. in the input, which misses encodings, symlinks, and whatever is invented next.

---

Testing

Use MemoryDriver. It is held to exactly the contract the real drivers are, including pagination:

from sillo.storage import Bucket, MemoryDriver, Public

bucket = Bucket("test", MemoryDriver(), policy=Public())
await bucket.put("a.txt", chunks(b"hello"), signed=True)

Or configure a whole application against it:

setup_storage(app, StorageConfig(
    default="scratch",
    buckets={"scratch": BucketConfig(driver="memory", policy=Public())},
))

signed=True bypasses the policy the way a signed URL would, which is how a worker or a fixture writes into a bucket whose policy has no user to ask about.

---

Writing a driver

Six methods, and a suite that tells you when you are finished:

from sillo.storage.base import Driver, FileInfo, Page, Stored

class AzureDriver(Driver):
    name = "azure"

    async def write(self, key, stream, *, content_type="", declared_type=""): ...
    def read(self, key): ...            # returns an async iterator
    async def stat(self, key): ...
    async def delete(self, key): ...
    async def page(self, prefix="", *, cursor="", limit=100): ...
    async def close(self): ...

exists, copy, move and capabilities have working defaults. Override copy if your backend can do it server-side, and signed_url if it can sign.

Then run the same battery every shipped driver passes:

from tests.test_storage.contract import DriverContract

class TestAzureDriver(DriverContract):
    @pytest.fixture
    async def driver(self):
        built = AzureDriver(...)
        yield built
        await built.close()

Thirty assertions, covering the edges that differ between backends: zero-byte objects, unicode keys, keys with spaces, paging across a boundary, overwriting, deleting what is not there, and common prefixes. If it passes, your driver is a drop-in for every other.

---

What this does not do

Deliberately, and unlikely to change:

• Image transformation and thumbnails. A different product.
• Virus scanning. Belongs in a queued job with a real scanner.
• CDN invalidation. Your CDN’s API, not a storage abstraction’s.
• A public_url() convenience. Every one of those becomes a bucket that is world-readable because somebody wanted a quick link.

See also

• Storage internals — the driver contract, the design decisions, and where “S3-compatible” leaks
• File uploads — parsing multipart before it reaches a bucket