Creating Bucket (S3) Notifications on AWS with Python

Cloud solutions with Nitric + Python (4 Part Series)

1 Your first Nitric API using Python
2 Creating Bucket (S3) Notifications on AWS with Python
3 Running a scheduled task with Nitric in Python
4 Secure Upload URLs Buckets with Nitric in Python

In this tutorial, we’ll explore how to set up bucket notifications in Python, enabling your applications to react to changes in file storage buckets. This is particularly useful in scenarios where you need to monitor a bucket for new file uploads or deletions.

If you haven’t used the Nitric SDK before, then start with this tutorial.

1. Create a Bucket Instance

Import the necessary packages and instantiate a new bucket object. In this tutorial, we’ll name our bucket ‘assets’.

<span>from</span> <span>nitric.resources</span> <span>import</span> <span>bucket</span>
<span>from</span> <span>nitric.application</span> <span>import</span> <span>Nitric</span>
<span>assets</span> <span>=</span> <span>bucket</span><span>(</span><span>"</span><span>assets</span><span>"</span><span>)</span>
<span>accessible_assets</span> <span>=</span> <span>bucket</span><span>(</span><span>"</span><span>assets</span><span>"</span><span>).</span><span>allow</span><span>(</span><span>"</span><span>reading</span><span>"</span><span>)</span>
<span>from</span> <span>nitric.resources</span> <span>import</span> <span>bucket</span>
<span>from</span> <span>nitric.application</span> <span>import</span> <span>Nitric</span>

<span>assets</span> <span>=</span> <span>bucket</span><span>(</span><span>"</span><span>assets</span><span>"</span><span>)</span>

<span>accessible_assets</span> <span>=</span> <span>bucket</span><span>(</span><span>"</span><span>assets</span><span>"</span><span>).</span><span>allow</span><span>(</span><span>"</span><span>reading</span><span>"</span><span>)</span>
from nitric.resources import bucket
from nitric.application import Nitric

assets = bucket("assets")

accessible_assets = bucket("assets").allow("reading")

Enter fullscreen mode Exit fullscreen mode

2. Set Up Notifications for File Write Events

To monitor for file write events, specifically for files starting with a certain prefix, use the On method on the bucket object. Let’s trigger notifications for files starting with ‘/users/images’.

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

<span>assets</span> <span>=</span> <span>bucket</span><span>(</span><span>"</span><span>assets</span><span>"</span><span>)</span>

<span>accessible_assets</span> <span>=</span> <span>bucket</span><span>(</span><span>"</span><span>assets</span><span>"</span><span>).</span><span>allow</span><span>(</span><span>"</span><span>reading</span><span>"</span><span>)</span>

<span># The request will contain the name of the file `key` and the type of event `type` </span><span>@assets.on</span><span>(</span><span>"</span><span>delete</span><span>"</span><span>,</span> <span>"</span><span>*</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_anything</span><span>(</span><span>ctx</span><span>):</span>
  <span>print</span><span>(</span><span>f</span><span>"</span><span>a file named </span><span>{</span><span>ctx</span><span>.</span><span>req</span><span>.</span><span>key</span><span>}</span><span> was deleted</span><span>"</span><span>)</span>

<span>@assets.on</span><span>(</span><span>"</span><span>write</span><span>"</span><span>,</span> <span>"</span><span>/images/cat</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_cat_image</span><span>(</span><span>ctx</span><span>):</span>
  <span>print</span><span>(</span><span>f</span><span>"</span><span>A cat image was written</span><span>"</span><span>)</span>

<span># If `on` is called with a permissioned bucket, a file will also be provided with the request </span><span>@accessible_assets.on</span><span>(</span><span>"</span><span>write</span><span>"</span><span>,</span> <span>"</span><span>/images/dog</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>access_dog_file</span><span>(</span><span>ctx</span><span>):</span>
  <span>dog_image</span> <span>=</span> <span>await</span> <span>ctx</span><span>.</span><span>req</span><span>.</span><span>file</span><span>.</span><span>read</span><span>()</span>

  <span>print</span><span>(</span><span>dog_image</span><span>)</span>

<span>Nitric</span><span>.</span><span>run</span><span>()</span>
from nitric.resources import bucket
from nitric.application import Nitric

assets = bucket("assets")

accessible_assets = bucket("assets").allow("reading")

# 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()

Enter fullscreen mode Exit fullscreen mode

3. Set Up Notifications for File Delete Events

Similarly, to monitor for file deletion events for any file, set up a notification like so:

<span>from</span> <span>nitric.resources</span> <span>import</span> <span>bucket</span>
<span>from</span> <span>nitric.application</span> <span>import</span> <span>Nitric</span>
<span># Create a reference to an 'assets' bucket with permissions to delete </span><span>assets</span> <span>=</span> <span>bucket</span><span>(</span><span>'</span><span>assets</span><span>'</span><span>).</span><span>allow</span><span>(</span><span>'</span><span>deleting</span><span>'</span><span>)</span>
<span>logo</span> <span>=</span> <span>assets</span><span>.</span><span>file</span><span>(</span><span>'</span><span>images/logo.png</span><span>'</span><span>)</span>
<span>await</span> <span>logo</span><span>.</span><span>delete</span><span>()</span>
<span>Nitric</span><span>.</span><span>run</span><span>()</span>
<span>from</span> <span>nitric.resources</span> <span>import</span> <span>bucket</span>
<span>from</span> <span>nitric.application</span> <span>import</span> <span>Nitric</span>

<span># Create a reference to an 'assets' bucket with permissions to delete </span><span>assets</span> <span>=</span> <span>bucket</span><span>(</span><span>'</span><span>assets</span><span>'</span><span>).</span><span>allow</span><span>(</span><span>'</span><span>deleting</span><span>'</span><span>)</span>

<span>logo</span> <span>=</span> <span>assets</span><span>.</span><span>file</span><span>(</span><span>'</span><span>images/logo.png</span><span>'</span><span>)</span>

<span>await</span> <span>logo</span><span>.</span><span>delete</span><span>()</span>

<span>Nitric</span><span>.</span><span>run</span><span>()</span>
from nitric.resources import bucket
from nitric.application import Nitric

# Create a reference to an 'assets' bucket with permissions to delete assets = bucket('assets').allow('deleting')

logo = assets.file('images/logo.png')

await logo.delete()

Nitric.run()

Enter fullscreen mode Exit fullscreen mode

Avoid writing or deleting to the bucket from within a notification as this can trigger the notification again, potentially causing an infinite loop which can be costly.

Let’s run this thing!

nitric start
nitric start
nitric start

Enter fullscreen mode Exit fullscreen mode

Conclusion

Now you have set up bucket notifications in Python using the Nitric SDK.

This setup will help you monitor file write and delete events, allowing your application to react to changes in your bucket accordingly.

Cloud solutions with Nitric + Python (4 Part Series)

1 Your first Nitric API using Python
2 Creating Bucket (S3) Notifications on AWS with Python
3 Running a scheduled task with Nitric in Python
4 Secure Upload URLs Buckets with Nitric in Python

原文链接：Creating Bucket (S3) Notifications on AWS with Python