Python - bucket.on()

Create a new bucket notification trigger when certain files are created or deleted.

from nitric.resources import bucket
from nitric.application import Nitric
assets = bucket("assets")
accessible_assets = bucket("assets").allow("read")
# The request will contain the name of the file `key` and the type of event `type`
@assets.on("delete", "*")
async def delete_anything(ctx):
print(f"a file named {ctx.req.key} was deleted")
@assets.on("write", "/images/cat")
async def create_cat_image(ctx):
print(f"A cat image was written")
# If `on` is called with a permissioned bucket, a file will also be provided with the request
@accessible_assets.on("write", "/images/dog")
async def access_dog_file(ctx):
dog_image = await ctx.req.file.read()
print(dog_image)
Nitric.run()

Earlier versions of the Nitric SDK used 'reading'. The latest version uses 'read'.

Parameters

  • Name
    notification type
    Required
    Required
    Type
    string
    Description

    The notification type for a triggered event, either on a file write or a file delete.

  • Name
    notification prefix filter
    Required
    Required
    Type
    string
    Description

    The file prefix filter that must match for a triggered event. If multiple filters overlap across notifications then an error will be thrown when registering the resource.

  • Name
    middleware
    Required
    Required
    Type
    BucketNotificationMiddleware
    Description

    The middleware (code) to be triggered by the bucket notification being triggered.

Available trigger types:

write

Run when a file in the bucket is created using: file.write()

delete

Run when a file in the bucket is deleted using: file.delete()

Trigger type cloud mapping

PermissionAWSGCPAzure
writes3:ObjectCreated:*OBJECT_FINALIZEMicrosoft.Storage.BlobCreated
deletes3:ObjectRemoved:*OBJECT_DELETEMicrosoft.Storage.BlobDeleted
Last updated on Oct 22, 2024