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

Permission	AWS	GCP	Azure
write	s3:ObjectCreated:*	OBJECT_FINALIZE	Microsoft.Storage.BlobCreated
delete	s3:ObjectRemoved:*	OBJECT_DELETE	Microsoft.Storage.BlobDeleted