Add GCSMediaStorage, and share a CloudMediaStorage base with S3

[regiscamimura]

Closes #333.

Adds GCSMediaStorage — and rather than a second copy of the S3 module, the shared object-storage plumbing moves into CloudMediaStorage, which S3MediaStorage now uses too.

from piccolo_api.media.gcs import GCSMediaStorage

MOVIE_POSTER_MEDIA = GCSMediaStorage(
    column=Movie.poster,
    bucket_name="my-bucket",
    folder_name="movie_posters",
)

Nothing else to wire up: Piccolo Admin's file field is driven by the MediaStorage interface, not by boto3 — so upload, preview, signed URLs and delete behave exactly as they do for S3.

CloudMediaStorage

The cloud SDKs are blocking, so every operation was written twice — a _sync method doing the work, and an async one running it in an executor. S3 and GCS had identical copies of that, plus __init__, the folder handling and __hash__/__eq__.

The base class owns all of it now. A backend implements the _sync methods plus upload_file:

class GCSMediaStorage(CloudMediaStorage):
    def upload_file(self, file_key, file, content_type): ...
    def generate_file_url_sync(self, file_key, root_url, user=None) -> str: ...
    def get_file_sync(self, file_key): ...
    def delete_file_sync(self, file_key): ...
    def bulk_delete_files_sync(self, file_keys): ...
    def get_file_keys_sync(self) -> list[str]: ...

S3MediaStorage's public API is unchanged — upload_metadata and its endpoint_url hashing stay in the subclass. Azure Blob would be a small module now, not a third copy.

Storage identity also moved up to MediaStorage, as a _hash_components() hook. MediaStorage defined __eq__ with no __hash__, which means Python sets __hash__ = None — so a custom backend written against the documented "subclass MediaStorage" path got a TypeError on == unless it knew to add __hash__ itself. Now it works by default, and LocalMediaStorage drops its hand-rolled pair.

Storage identity moved up to MediaStorage too, as a _hash_components() hook — Piccolo Admin uses set(media_storage) to catch two columns writing to the same place, and each backend now declares what "the same place" means for it. MediaStorage previously defined __eq__ with no __hash__, so a custom backend written against the documented "subclass MediaStorage" path hit a TypeError on == unless it knew to add __hash__ itself.

Bugs fixed in S3 along the way

Found while unifying the two backends. Each has a regression test, and I checked each one fails without the fix:

• get_file_keys() stripped the folder with str.lstrip(prefix) — that removes characters, not a prefix. A file called poster.jpg in a movie_posters folder came back as .jpg, so delete_unused_files() compared mangled keys against the database. The existing test passes only because its fixture is named bulb.jpg.
• bulk_delete_files() sent the entire key list on every iteration instead of the current batch, and (iteration + 1 * batch_size) doesn't group the way it reads. Over 1000 keys, S3 rejects the request.
• store_file_sync() wrote ContentType back into self.upload_metadata. It leaked into the next file with an unknown extension — and with concurrent uploads it could be applied to the wrong file entirely, since the executor runs 10 at a time.
• bulk_delete_files() never looked at the Errors array delete_objects returns. S3 reports a key it couldn't delete in the response body rather than raising, so cleaning a bucket without s3:DeleteObject returned quietly having deleted nothing.
• A folder_name with a trailing slash stored files under movie_posters/ but listed them under movie_posters//, so delete_unused_files() silently found nothing. _prepend_folder_name now derives from the same folder_prefix the listing uses, so they can't drift (this also removes a pathlib call that would emit backslashes on Windows).
• photo.JPG uploaded with no ContentType, because CONTENT_TYPE is keyed by lowercase extension while validate_file_name lowercases and this didn't. Browsers downloaded the image instead of showing it.

Happy to split these into a separate PR if you'd rather review them apart from the GCS work.

Testing

You mentioned the Azure PR stalled because it was hard to test, so GCS runs against a faked storage.Client — no live bucket, no credentials. piccolo_api/media is at 100% coverage.

Signed URLs on GCP compute took two goes to get right, so it's worth spelling out:

1. blob.generate_signed_url() only routes through the IAM signBlob API if you pass both service_account_email and access_token. Without them it wants a private key, which ADC from the metadata server doesn't have — so granting roles/iam.serviceAccountTokenCreator on its own does nothing, and every URL raises AttributeError.
2. Passing the client's own token isn't enough either. storage.Client asks google.auth.default for the three devstorage scopes, and — unlike Compute Engine — Cloud Run's metadata server honours the scopes it's asked for. IAM signBlob only accepts cloud-platform, so that token gets a 403.

So the credentials are re-scoped to cloud-platform before signing, and cached. A service-account JSON still signs locally and never touches IAM.

I can't test this against live GCP — the branch logic is covered by fakes, but if anyone can try it on Cloud Run, that's the bit worth checking.

The GCS fake deliberately behaves like the real client where it matters: it defers operations inside a batch, distinguishes a missing object from a permissions error, and reports the first failure after applying every operation. So the batching test fails if the code isn't really batching, rather than passing regardless.

Also worth noting: CI didn't install the gcs extra, so tests/media/test_gcs.py aborted collection and took the whole run with it. Fixed in all three jobs.

Consistency fixes across all three backends

• Both cloud backends cache their client (building one resolves credentials — on Cloud Run that's a metadata round trip per file on a list page). The caches are locked, so a cold cache under concurrent requests doesn't build one per thread.
• GCS deletes use delete_blobs(on_error=...), not client.batch(). Batching would be 100× fewer requests, but a batch reports only one of its failures and Batch._finish_futures keeps the last one (a 4xx Response is falsy, so exception_args or subresponse walks past earlier errors) — so a permissions error can hide behind an already-deleted key and the sweep looks like it worked. For a delete, silently skipping files is worse than being slow.
• bulk_delete_files tolerates a file that has already gone on every backend, so one stale key can't abandon the sweep. Anything else — a permissions error, say — still raises. A single delete still raises if the file is missing; naming one file means you want to know.
• LocalMediaStorage follows the same convention as the cloud ones now (_sync does the work, async wraps it). Its bulk_delete_files and get_file_keys were doing filesystem work directly in the event loop.

Known limitation

On GCP compute, each signed URL costs an IAM signBlob request, because ADC can't sign locally. Documented in the docstring, with the two ways out (mount a key, or sign_urls=False with a public bucket). Client caching doesn't help with this part.

get_file_keys() also lists objects in nested folders under the prefix, which delete_unused_files() will then remove since they never match a database value. That behaviour is unchanged from master — flagging it rather than changing it, since a shared bucket is arguably a misconfiguration and the fix would be a semantic change.