DEV Community: Ademola Akinsola

Logging in Python: A Beginner's Guide

Ademola Akinsola — Thu, 02 Feb 2023 06:00:39 +0000

Logging is an important aspect of any application. It is the process of providing information about the happenings in your application. If done well, It can be useful for helping a developer pinpoint reasons for app crashes, errors, possible bugs, and areas of poor performance faster.

In this article, we will learn how we can add logging into our python application. To follow this article, basic knowledge of python would be useful.

Logging components

At a high level, logging in python has three main components. These are:

Loggers : These are python objects that allow us to define the log messages.
Handlers : These are objects that can be attached to a logger. They decide how to process a log.
Formatters : These are objects that can be attached to a handler. They help to format the way logs are displayed.

Loggers

Python makes it easy to quickly get started with logging. Let us create an app.py file to see how it works.

import logging

LOGGER = logging.getLogger('foo.bar')

LOGGER.debug('Nothing to see here, Just trying to Debug')
LOGGER.info('Hey, here is some information you might find useful')
LOGGER.warning('something may be wrong here!!! warning')
LOGGER.error('An error definetely just happened. You might want to take a look!')
LOGGER.critical('Oops.. Critical server error.')

If we execute the file with python3 app.py on our terminal, we will get the below in our terminal.

something may be wrong here!!! warning
An error definetely just happened. You might want to take a look!
Oops.. Critical server error.

We use the logging module provided by python to create a Logger object and log messages onto our terminal.

But wait🤨, we only see three of the five logs we wrote, what's happening? Well, this is because of how logs are propagated. Python provides us with a default root logger which is the topmost logger. By default, loggers in python follow a hierarchical order, where logs are propagated upwards to their parents till the log reaches the root logger.

A logger named foo.bar.baz would be a child of a logger created at foo.bar and foo, and hence, will send logs upwards to them.

Now, the root logger has what we call an effective level of "warning". An effective level is what a logger uses to decide if it will consider processing a log message. If a logger has an effective level of "warning", it will only consider processing logs of priority "warning" and above, a log message of priority "debug" or "info" will be ignored. This is exactly why we only saw three of our five logs being printed above.

By default, we have the following priority levels provided.

All loggers, except the root logger, are created with an effective level NOTSET. An effective level of NOTSET means the logger will delegate the decision of logging this message to its parent logger.

If we would like to log messages with a priority lower than that of our root logger's effective level. We need to explicitly create a handler for our logger.

Handlers

Handlers assist in the decision-making of whether to process a log message based on the log's priority and its own specified priority level. To create one, edit your app.py file.

import logging

LOGGER = logging.getLogger('foo.bar')

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)

LOGGER.addHandler(stream_handler)
LOGGER.setLevel(logging.DEBUG)

LOGGER.debug('Nothing to see here, Just trying to debug')
LOGGER.info('Hey, here is some information you might find useful')
LOGGER.warning('something may be wrong here!!! warning')
LOGGER.error('An error definitely just happened. You might want to take a look!')
LOGGER.critical('Oops.. Critical server error.')

We added a handler named StreamHandler to our logger, this handler will output the logs it receives to the terminal (by default, stderr). We set its level to DEBUG to allow it to process messages of priority "DEBUG" and higher. We also set the effective level of the logger object to "DEBUG", since, if you recall, all loggers (except root) have a default effective level of "NOTSET" making them unable to act on a log by themselves, and will propagate the log upwards.

Now if we execute our app.py script

Nothing to see here, Just trying to debug
Hey, here is some information you might find useful
something may be wrong here!!! warning
An error definitely just happened. You might want to take a look!
Oops.. Critical server error.

Great! 😎 We see all our logs now. You can tweak the priority levels of both the logger and a handler to get your desired output.

Using a FileHandler

We can also have multiple handlers on a single logger. Another common handler is the FileHandler. Let's see it in action, update your app.py

import logging

LOGGER = logging.getLogger('foo.bar')

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)

LOGGER.addHandler(stream_handler)
LOGGER.setLevel(logging.DEBUG)
# ------ new lines ---------
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)
LOGGER.addHandler(file_handler)
# --------------------------

LOGGER.debug('Nothing to see here, Just trying to debug')
LOGGER.info('Hey, here is some information you might find useful')
LOGGER.warning('something may be wrong here!!! warning')
LOGGER.error('An error definetely just happened. You might want to take a look!')
LOGGER.critical('Oops.. Critical server error.')

Execute the script and you should see a new file named app.log appear in your working directory. Inspecting it, you should see the logs written into it.

It is common practice to add multiple handlers with different priority levels to a handler. You might want all logs printed into the terminal using a StreamHandler but only want error and critical logs written to your log files. This can easily be achieved by setting the StreamHandler level to DEBUG and the FIleHandler's level to "ERROR".

Integrating the RotatingFileHandler

For large services with lots of logs being emitted, log files tend to get very large very quickly and this can be a problem as it gets harder to manage and can end up taking up a lot of the space on the machine. To solve this, we use a RotatingFileHandler. After our log file reaches a specified size, it will stop logging into the file and rename it usually appending an increasing integer value at the end of the file name, then creating an empty log file of the original name and resuming logging into that. So you will tend to see log files like app.log, app.log.1, app.log.2, app.log.3

Update the app.py file

import logging
from logging.handlers import RotatingFileHandler # new

LOGGER = logging.getLogger('foo.bar')

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)

LOGGER.addHandler(stream_handler)
LOGGER.setLevel(logging.DEBUG)

file_handler = RotatingFileHandler('app.log', maxBytes=100, backupCount=5) # updated
file_handler.setLevel(logging.DEBUG)
LOGGER.addHandler(file_handler)

LOGGER.debug('Nothing to see here, Just trying to debug')
LOGGER.info('Hey, here is some information you might find useful')
LOGGER.warning('something may be wrong here!!! warning')
LOGGER.error('An error definetely just happened. You might want to take a look!')
LOGGER.critical('Oops.. Critical server error.')

We imported the RotatingFileHandler class and then instantiated it with the file name and two arguments

maxBytes : specifies the size the file must reach before we carry out a log rotation. We use a low value here to easily see the rotation in action, usually, it would be much higher, edit it to suit your needs.
backupCount : specifies the number of logs to keep before beginning to delete the oldest logs.

Execute the app.py script, you should see about two new log files in your working directory.

When app.log got filled, it was renamed as app.log.1, then a new app.log was made, when that new app.log got filled, it became app.log.1 and the previous app.log.1 became app.log.2.

Formatters

Formatters are objects that can be attached to a handler. They define the way logs should be displayed. Let us take an example

import logging
from logging.handlers import RotatingFileHandler

LOGGER = logging.getLogger('foo.bar')

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.DEBUG)

LOGGER.addHandler(stream_handler)
LOGGER.setLevel(logging.DEBUG)

file_handler = RotatingFileHandler('app.log', maxBytes=100, backupCount=5)
file_handler.setLevel(logging.DEBUG)
LOGGER.addHandler(file_handler)

# ------ new lines --------
formatter = logging.Formatter(fmt='%(asctime)s: %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
file_handler.setFormatter(formatter)
stream_handler.setFormatter(formatter)

LOGGER.debug('Nothing to see here, Just trying to debug')
LOGGER.info('Hey, here is some information you might find useful')
LOGGER.warning('something may be wrong here!!! warning')
LOGGER.error('An error definetely just happened. You might want to take a look!')
LOGGER.critical('Oops.. Critical server error.')

We created a formatter and specified how to display our logs. We then attached it to the file and stream handler. Let us execute the script to see the result.

2023-01-31 00:44:08: foo.bar - DEBUG - Nothing to see here, Just trying to Debug
2023-01-31 00:44:08: foo.bar - INFO - Hey, here is some information you might find useful
2023-01-31 00:44:08: foo.bar - WARNING - something may be wrong here!!! warning
2023-01-31 00:44:08: foo.bar - ERROR - An error definetely just happened. You might want to take a look!
2023-01-31 00:44:08: foo.bar - CRITICAL - Oops.. Critical server error

As you can see, new information has been added to our logs and it resembles the structure we defined for the formatter with the fmt argument.

Tidy up logging with configurations

Python provides us with several ways of configuring all or most of our loggers in one place. By providing a dictionary configuration we can specify which loggers should exist, the handlers to be attached, as well as formatters. Let us see how this is done.

Logging with DictConfig

In a new file called logging_config.py paste the following code

LOGGING_CONFIG = { 
    'version': 1,
    'disable_existing_loggers': True,
    'formatters': { 
        'basic': {
            'format': '%(asctime)s: %(name)s - %(levelname)s - %(message)s',
            'datefmt' : '%Y-%m-%d %H:%M:%S'
        },
    },
    'handlers': { 
        'stream_handler': { 
            'level': 'DEBUG',
            'formatter': 'basic',
            'class': 'logging.StreamHandler',
        },
        'rotating_file_handler': { 
            'level': 'ERROR',
            'formatter': 'basic',
            'class': 'logging.handlers.RotatingFileHandler',
            'backupCount': 5,
            'maxBytes': 100,
            'filename': 'app.log'
        },
    },
    'loggers': { 
        '': {
            'handlers': ['stream_handler'],
            'level': 'WARNING',
            'propagate': False
        },
        'app': { 
            'handlers': ['stream_handler', 'rotating_file_handler'],
            'level': 'DEBUG',
            'propagate': False
        },
    } 
}

The config file is very similar to what we have been doing with logging helper methods like setFormatter and addHandler. Dissecting this config dictionary:

We have one formatter here defined, named as " basic". We specify our log format with the format key and our date format with the datefmt key.
We define two handlers, stream_handler and rotating_file_handler, similar to what we have done before. We set the priority level of "DEBUG" for the stream handler and "ERROR" for the rotating file handler.
Finally, we define our loggers. The empty string key is synonymous with saying 'root'. We give our root logger the stream_handler handler we defined earlier. Then define a new logger named app and give it both handlers.
We set propagate key to false since we explicitly defined handlers for our loggers and we don't want duplicated logging from loggers higher up on the hierarchy.

Now in our app.py file

import logging
import logging.config
from logger_config import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)

LOGGER = logging.getLogger('app')

LOGGER.debug('Nothing to see here, Just trying to debug')
LOGGER.info('Hey, here is some information you might find useful')
LOGGER.warning('something may be wrong here!!! warning')
LOGGER.error('An error definetely just happened. You might want to take a look!')
LOGGER.critical('Oops.. Critical server error.')

As you can see, our app file looks a lot cleaner. Now, run the script again.

You should notice new logs in both the terminal and the log files. If you look closely, you should realize all logs are displayed on the console, but only "ERROR" and "CRITICAL" logs are shown in our log file. This is because of how we defined our LOGGING_CONFIG, "stream_handler" has its effective level set to "DEBUG" and "rotating_file_handler" has its effective level as "ERROR".

...And that's it.

Conclusion

In this article, we learned the importance of logging in python, the components of logging in python, and how to create and use our own loggers. We also saw how to specify new formats for our logs, and how to send logs to different outputs like console/terminal and files.

If you found this article useful or learned something new, consider dropping a heart up and following me to keep up-to-date with any recent postings!

You can also find me on Twitter at akinsola232 and on LinkedIn at Ademola.

Till next time, happy coding!

Signed.... Humanity's Strongest Soldier, Levi🧑💻👋.

How to download data in multiple file formats (CSV, XLS, TXT) with Django REST Framework

Ademola Akinsola — Wed, 25 Jan 2023 12:29:19 +0000

A lot of the time, our typical server response is in the form of JSON or XML. This serves our use cases a good number of times, however, there are times when the need to provide data in the form of a file arises.

In this article, we will be exploring how to convert our model data into files and send them as responses in Django REST Framework (DRF). We will do this by building out a simple project.

Prerequisites

You should have some basic Django and Django REST Framework knowledge. You should also have python already installed on your system.

The project, Setting up and Installation

We will be building out a simple Student Management App. The purpose of the app will be to:

Allow a user to input student data
Download all the available data as either a CSV, Excel or TXT file.

Creating the project and Installing Dependencies

We will install Django and DRF to start. To do this, first, create your virtual environment and activate it.

python3 -m venv studentappenv
source studentappenv/bin/activate

Now install Django and DRF. We will also be installing openpyxl which is a python package that helps when interacting with data and excel files.

pip install django djangorestframework openpyxl

Next, create the Django project and create an app.

django_admin startproject student_management .
python manage.py startapp student_data

Update your settings.py and include the new app inside the INSTALLED_APPS list.

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'student_data', #new line
]

Let's create our model. Head over to student_data/models.py

from django.db import models

class StudentData(models.Model):

    STUDENT_GENDER = ((1, 'Male'), (2, 'Female'), (3, 'Other'))
    STUDENT_LEVEL = ((1, 'Junior'), (2, 'Senior'))

    first_name = models.CharField(max_length=100)
    last_name = models.CharField(max_length=100)
    age = models.IntegerField()
    gender = models.IntegerField(choices=STUDENT_GENDER)
    level = models.IntegerField(choices=STUDENT_LEVEL)

A pretty simple model that stores basic student information. This will be good enough for our needs.

Next up, we create our serializer. Create a new file serializers.py inside the student_data folder and add the following inside.

from rest_framework import serializers
from .models import StudentData

class StudentDataSerializer(serializers.ModelSerializer):

    class Meta:
        model = StudentData
        fields = ' __all__'

    def to_representation(self, instance):
        data = super().to_representation(instance)

        data['gender'] = instance.get_gender_display()
        data['level'] = instance.get_level_display()

        return data

This is also a basic serializer for our StudentData model. We override our to_representation to change how the server would return values for the gender and level fields. Instead of returning something like 'gender': '1', we will instead get 'gender': 'Male'.

We have our model and serializer, now, let's create our view. Inside student_data/views.py, add the following lines of code:

from django.utils import timezone
from rest_framework.response import Response

from rest_framework.viewsets import ModelViewSet
from rest_framework.decorators import action
from .models import StudentData
from .serializers import StudentDataSerializer

class StudentDataViewset(ModelViewSet):
    serializer_class = StudentDataSerializer
    queryset = StudentData.objects.all()

    @action(detail=False, methods=["get"])
    def download(self, request):
        queryset = self.get_queryset()
        serializer = StudentDataSerializer(queryset, many=True)

        return Response(serializer.data)

This is our view which will handle POST requests to create our student data. We have also defined a new viewset action download which we will use for downloading our file responses. Right now it will simply return the data as a JSON response, we will tweak it soon to meet our needs.

Let's connect our URL paths. Create a urls.py file in the student_data folder.

# student_data/urls.py

from .views import StudentDataViewset
from django.urls import path, include

from rest_framework import routers

router = routers.DefaultRouter()
router.register("", StudentDataViewset, basename="student-data")

urlpatterns = [
    path("", include(router.urls)),
]

Connect this to the main urls file at student_management folder.

# student_management/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('student-data/', include('student_data.urls')),
]

Run makemigrations and migrate

python manage.py makemigrations
python manage.py migrate

Start the server with python manage.py runserver.

Make a few POST requests at http://127.0.0.1:8000/student-data/ to populate our database with some data.

Now test the http://127.0.0.1:8000/student-data/download/ endpoint with a GET request to see the response type. You will get JSON data as a response similar to what we saw from our POST request. This is because DRF sets a global default renderer rest_framework.renderers.JSONRenderer which will return server responses in application/json media type.

We will create our own renderers which will convert our data into the response formats we want.

Create a renderers.py file inside the student_data folder.

# student_data/renderers.py
import io
import csv
from rest_framework import renderers

STUDENT_DATA_FILE_HEADERS = ["id", "first_name", "last_name", "age", "gender", "level"]

class CSVStudentDataRenderer(renderers.BaseRenderer):

    media_type = "text/csv"
    format = "csv"

    def render(self, data, accepted_media_type=None, renderer_context=None):

        csv_buffer = io.StringIO()
        csv_writer = csv.DictWriter(csv_buffer, fieldnames=STUDENT_DATA_FILE_HEADERS, extrasaction="ignore")
        csv_writer.writeheader()

        for student_data in data:
            csv_writer.writerow(student_data)

        return csv_buffer.getvalue()

This is going to be the renderer that will convert our data into CSV format. Let's break this code down.

We set a STUDENT_DATA_FILE_HEADERS list which contains all our model fields.
Our CSVStudentDataRendererclass subclasses BaseRenderer and defines the media_type and format attributes. We have many different media types which you can find here at IANA defined media types (mime types).
We define a render method where a bulk of the work is done. Inside it:

We create a string buffer named csv_buffer.
We create an instance of csv.DictWriter and pass in our buffer as an argument.
We call csv_writer.writeheader() which will write the values in STUDENT_DATA_FILE_HEADERS as the first line of our CSV file.
We use the csv helper module to write our data as comma separated values in the buffer location.
We retrieve the content written into our buffer with csv_buffer.getvalue()

Let's add renderers for Excel and Text files next. Update renderers.py file.

import io
import csv
import openpyxl # new
from rest_framework import renderers

STUDENT_DATA_FILE_HEADERS = ["id", "first_name", "last_name", "age", "gender", "level"]

...

# ------ NEW LINES --------
class TextStudentDataRenderer(renderers.BaseRenderer):

    media_type = "text/plain"
    format = "txt"

    def render(self, data, accepted_media_type=None, renderer_context=None):

        text_buffer = io.StringIO()
        text_buffer.write(' '.join(header for header in STUDENT_DATA_FILE_HEADERS) + '\n')

        for student_data in data:
            text_buffer.write(' '.join(str(sd) for sd in list(student_data.values())) + '\n')

        return text_buffer.getvalue()

class ExcelStudentDataRenderer(renderers.BaseRenderer):

    media_type = "application/vnd.ms-excel"
    format = "xls"

    def render(self, data, accepted_media_type=None, renderer_context=None):    

        workbook = openpyxl.Workbook()
        buffer = io.BytesIO()
        worksheet = workbook.active
        worksheet.append(STUDENT_DATA_FILE_HEADERS)

        for student_data in data:
            worksheet.append(list(student_data.values()))

        workbook.save(buffer)

        return buffer.getvalue()

The idea is the same as our CSVStudentDataRenderer. We define the right media_type and format of our expected file. We then create a buffer which we write our headers and model data into. After that, we return the content of the buffer. In the case of our ExcelRenderer, we use the openpyxl library to simplify the buffer writing process.

Now that we have our renderers, let's update our view to use it. Update your student_data/views.py file.

from django.utils import timezone
from rest_framework.response import Response

from rest_framework.viewsets import ModelViewSet
from rest_framework.decorators import action
from .models import StudentData
from .renderers import CSVStudentDataRenderer, ExcelStudentDataRenderer, TextStudentDataRenderer
from .serializers import StudentDataSerializer

class StudentDataViewset(ModelViewSet):
    serializer_class = StudentDataSerializer
    queryset = StudentData.objects.all()

    @action(detail=False, methods=["get"], renderer_classes=[CSVStudentDataRenderer, ExcelStudentDataRenderer, TextStudentDataRenderer])
    def download(self, request):
        queryset = self.get_queryset()

        now = timezone.now()        
        file_name = f"student_data_archive_{now:%Y-%m-%d_%H-%M-%S}.{request.accepted_renderer.format}"
        serializer = StudentDataSerializer(queryset, many=True)
        return Response(serializer.data, headers={"Content-Disposition": f'attachment; filename="{file_name}"'})

The three main takeaways from the new changes are:

We define a renderers_classes parameter in our @action decorator. This will be used by the download endpoint when choosing a renderer. How will it decide? we will talk about that in a minute.
We create a distinct file name based on the current time and append the format of the renderer which would be used to serve the particular response, found at request.accepted_renderer.format.
We set a content-dispostion header which lets the client know that the response should be treated as an attachment that should be downloaded under the value of filename.

Let's head back to postman to test our download endpoint.

I get a CSV file downloaded onto my system.

Great. This means Django decided to use our CSVRenderer, but how did it decide? Django has a content negotiation mechanism which it uses to determine how to render a response back to the client. It looks into the Accept header in the request and tries to map it to an available renderer. If it is unable to map it to any specific one, by default, Django chooses the first renderer in the renderers_classes list. This is why we get a CSV file in return.

Let's edit our Accept request header and specify the Excel media type(application/vnd.ms-excel) then retry the request.

I now get an Excel file downloaded.

And if we update the Accept header to text/plain we get a txt file downloaded as Django chooses the text renderer class.

That's it.

Conclusion

In this article, we learned how to create our own custom renderers for specific media types and formats like excel, CSV and text. We also saw how to help our application pick the right renderer using the Accept request header.

If you found this article useful or learned something new, consider dropping a heart up and following me to keep up-to-date with any recent postings!

You can also find me on Twitter at akinsola232 and on LinkedIn at Ademola.

Till next time, happy coding!

Levi

Introduction to Elastic Beanstalk with Django, RDS, Docker and Nginx

Ademola Akinsola — Thu, 19 Jan 2023 17:37:47 +0000

In this article, we will learn about Elastic Beanstalk and its capabilities. We will understand the problem that Elastic Beanstalk solves and the steps required to set up and deploy a Django app with Elastic Beanstalk.

We will also be using Docker for containerizing the app and Nginx as a reverse proxy server. Additionally, we will learn how to connect an RDS instance with our Beanstalk application.

What is Elastic Beanstalk?

Elastic Beanstalk (EB) is a managed AWS service that allows you to upload, deploy and manage your applications easily. It deals with the provisioning of resources needed for the application such as EC2 instances, Cloudwatch for logs, Auto Scaling Groups, Load Balancers, Databases, and Proxy Servers (Nginx, Apache) - all of which are customizable.

Setting up the bare project

To focus on the main theme of this article, Elastic Beanstalk, we will be cloning a very simple Django app where we make a post request to add some inventory data into our database and a get request to retrieve them. Head over to this github repository and clone the app.

Setting up Docker

We will be using docker for containerization. This is to maintain consistency between the development environment and production environments and eliminate all forms of But it works on my computer problems.

For this, you should have docker and docker-compose already installed.

Writing our Dockerfile

Let us create a Dokcerfile in the root directory. Copy the following into it.

FROM python:3.9-bullseye

ARG PROJECT_DIR=/home/app/code
ENV PYTHONUNBUFFERED 1

WORKDIR $PROJECT_DIR

RUN useradd non_root && chown -R non_root $PROJECT_DIR

RUN python -m pip install --upgrade pip
COPY requirements.txt $PROJECT_DIR
RUN pip3 install --no-cache-dir -r requirements.txt
COPY . $PROJECT_DIR
RUN python manage.py collectstatic --noinput

EXPOSE 8000

USER non_root

What are we doing here?

We specify our base image as python:3.9-bullseye with the FROM. This is, to put it simply, a Debian OS with python3.9 installed.
We create a new directory where we will be moving our Django application code into.
We create a new user and give it ownership of the application code directory (This is to follow docker security best practices).
We install the app dependencies from requirements.txt and run collectstatic to collect all our static files into a single directory path which we have already defined in our settings.py file as STATIC_ROOT.
We EXPOSE port 8000 of the container. The proxy server we will create will forward requests to this port.

Writing our docker-compose file

Now, Let us add a compose file. Create a docker-compose.yml file at the root directory and add the following.

version: '3.6'
services:
  web:
    build:
      context: .
    command: sh -c "python manage.py migrate &&
                    gunicorn ebs_django.wsgi:application --bind 0.0.0.0:8000" 
    volumes: 
      - static_volume:/home/app/code/static
    env_file: 
      - .env
    image: web
    restart: "on-failure"
  nginx:
    build: 
      context: ./nginx
    ports: 
    - 80:80 
    volumes:  
      - static_volume:/home/app/code/static 
    depends_on:
      - web 

volumes:
  static_volume:

We define two services here:

WEB: This is our application. It will be built based on instructions defined in our Dockerfile.
Once built, we run manage.py migrate to create our database schemas from our migration files. Then we bind gunicorn to the port 8000 of the machine. Gunicorn will serve our Django app through it.
NGINX: This is a web server we will be using as our reverse proxy. (A reverse proxy server is a server that sits in front of our application and routes incoming requests from external sources (users) to our provisioned servers.)
We will also use Nginx to serve our static files. hence the reason for binding to the volume static_volume which will be populated by our web container when we run collectstatic.

Next, create a folder named Nginx and add a Dockerfille inside with the following:

FROM nginx:1.23.3

RUN rm /etc/nginx/conf.d/default.conf
COPY nginx.conf /etc/nginx/conf.d/nginx.conf

This is the Dockerfile our Nginx container will be created from. It just defines a base image and deletes the default.conf provided by Nginx and adding a new one named nginx.conf, which we will create now.

Create the file and copy the following into it:

server {

    listen 80 default_server; # default external port. Anything coming from port 80 will go through NGINX

    location / {
        proxy_pass http://web:8000;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header Host $host;
        proxy_redirect off;
    }
    location /static/ {
        alias /home/app/code/static/; # Our static files
    }
}

What is happening here?

The Nginx server listens on the machine's port 80 for incoming requests.
Requests received are forwarded to our Django (gunicorn) server.
We define another location directive /static/, this is so that Nginx directly serves our static files as we mentioned earlier.

NOTE: We manually set up Nginx as a container since EB does not create an Nginx proxy server for us when we use a docker-compose.yml file, if we were to use a Dockerfile only, EB would provision an Nginx server for us.

Installing necessary dependencies

We need to install some packages for our docker container. Since we will be using gunicorn to serve our Django app and we will be using PostgreSQL, we need to install psycopg2-binary as well as gunicorn. You can do that with the following command in your activated virtual environment on the shell:

(ebsenv) ebs-django-docker-tutorial git:(master) pip install psycopg2-binary gunicorn

Run pip freeze to update your requirements.txt file.

(ebsenv) ebs-django-docker-tutorial git:(master) pip freeze > requirements.txt

Integrating Elastic Beanstalk

First off, let's understand two terms:

Environment: This is a collection of all the AWS resources provisioned by EB to run your application code.
Application: This is a collection of environments, application code versions, and environment configurations. You can have multiple environments in a single application. For example, you could decide to have a production environment running a specific application code version, and a QA environment running another application code version.

Alright, now, back to the code. EB provides us with several tools we can use to configure our environments and deploy our applications.

EB CLI
AWS SDK (Boto for python)
EB Console

We will be using a mix of both the CLI and the EB console. To install the cli, use the command below inside your virtual env.

(eb_python_testenv) EBS-TEST git:(master) pip install awsebcli

Head over to the AWS console to get your AWS credentials (aws_access_key, aws_secret_access_key), rename .env.example file to .env and paste your credentials there.

We can run eb init to initialize some configurations (platform, region) for our application.

(eb_python_testenv) EBS-TEST git:(master) eb init eb-docker-rds-django

By default, EB provisions the resources in your application inside the us-west-2 (US West (Oregon)) region, you can choose a different region by picking its corresponding number in the CLI. For this tutorial, we will be using the us-east-1. It asks if you are using Docker, choose Yes, Then Pick the option to use Docker when asked to select a platform. When prompted to use CodeCommit, pick no. CodeCommit is a Version Control Service and we are already using GitHub. When asked to set up SSH, choose no as we will not be needing it here.

Once that's done, you will see a new .elasticbeanstalk folder created for you with a config.yml file created. It defines some configurations for when we eventually create our environment.

Commit your changes with git add . and git commit -m "your commit message"

Next, we create our environment with the command:

(ebsenv) ebs-django-docker-tutorial git:(master) eb create eb-docker-rds-django

eb-docker-rds-django is our environment name. Wait for a couple minutes while EB provisions your resources for you. Once that is done, We can head over to the AWS EB console to see the status of our environment or we could use eb status on our terminal directly.

As you can see currently, the Health of our app is red. This is because in our settings.py file we are attempting to retrieve credentials for a database we do not have yet. You can confirm this by typing in eb logs on your terminal and scrolling through the logs.

Let us create an AWS Relational Database Service (RDS) for our EB environment.

Head over to AWS RDS console, click on databases and click Create database . Next,

Choose Standard create option.
Choose PostgreSQL as your database engine.
Under Credentials settings, type in a master username and a master password. (Write them down somewhere).
Scroll to additional configurations, expand, and under Initial database name, type the name you want to use for your application database (Write it down).
Leave everything else the same and click Create Database .

Once that is created successfully, we need to edit the security groups to let the EC2 instances of our EB environment access our RDS instance. To do this:

Click on the newly created database. Scroll down to Security group rules and click on the Inbound Security Group. Click on Inbound Rules tab and click on Edit inbound rules.
For type, choose Postgres.
For the source, we will choose the security group attached to our EC2 instances by EB. To find the security group of your EC2 instances, you can simply head to Auto Scaling Group (ASG) Service, click on your provisioned ASG, scroll to Launch Configurations and click on the security group, on the new page, you will see the security group id of your ASG, copy it and paste it into the source search bar.
Click save rules.

We have created our database, now we need to add the environment variables inside our EB environment. EB provides us several ways to do that, including using the CLI with eb setenv key=value or directly through the console. Let us head over to the EB console to do this. Under environments, click on your environment. On the left sidebar, click on configuration, Under Software click Edit and scroll down to Environment properties, and add your RDS credentials. Also, add your DJANGO_SECRET_KEY .

You can find your database hostname under the connectivity tab of the RDS instance.

Next, we need to add our domain into the allowed_hosts settings of Django. Head over to the EB console, click on your environment and grab the URL.

Go to your settings.py file and update the ALLOWED_HOSTS list.

# ebs_django/settings.py
# looks something like eb-docker-rds-django.xxx-xxxxxxxx.us-east-1.elasticbeanstalk.com
ALLOWED_HOSTS = ['YOUR_ENVIRONMENT_URL']

commit your changes with git add . and git commit -m "your commit message" and deploy the new version with:

(ebsenv) ebs-django-docker-tutorial git:(master) eb deploy

That's it. Head over to postman to test the app.

It works, yay! I guess we're done... well, not quite. The EB CLI provides us with the tools necessary to run our environment locally, which usually would be great when doing development. We can do this with the command eb local run. Unfortunately, EB "local run" currently has no support for reading docker-compose.yml files, so what's a workaround?

Docker Compose for Development

We can simply create a new docker-compose file to simulate our EB environment. Luckily, in our current setup, we really just need to add a database and we are set. Let's do that.

Create a new docker-compose file called docker-compose.dev.yml . This will be used specifically for development purposes.

version: '3.6'
services:
  web:
    build: 
      context: .
    command: sh -c "python manage.py makemigrations && python manage.py migrate &&
                    gunicorn ebs_django.wsgi:application --bind 0.0.0.0:8000 --reload" # updated command
    volumes: 
      - ./:/home/app/code/ # new volume
      - static_volume:/home/app/code/static
    env_file: 
      - .env
    image: web
    depends_on:
      - db
    restart: "on-failure"

  nginx:
    build: 
      context: ./nginx
    ports: 
    - 80:80 
    volumes:  
      - static_volume:/home/app/code/static 
    depends_on:
      - web
  db: # new db service
    image: postgres:15-alpine 
    volumes:
      - pgdata:/var/lib/postgresql/data
    environment:
      - POSTGRES_PASSWORD=${RDS_PASSWORD}
      - POSTGRES_DB=${RDS_DB_NAME}
      - POSTGRES_USER=${RDS_USERNAME}
    ports:
      - 5432:5432

volumes:
  pgdata: # new line
  static_volume:

There are a few differences in this compose file compared with our initial one.

the command in our web service now includes a reload flag. This ensures gunicorn reloads the server after new source code changes
A new volume in web to make changes in our local files reflect in our container.
A new db service.
A new pgdata volume to make sure our db data persists.

Update your .env file to include the credentials for our development DB.

# Develop
DJANGO_SECRET_KEY='secret_key'
DEBUG="1"
ALLOWED_HOSTS=*

# --- ADDED LINES -----
RDS_HOSTNAME=db
RDS_DB_NAME='test'
RDS_PASSWORD='test123'
RDS_USERNAME='test'
RDS_PORT=5432
# ---------------------
aws_access_key=AKIA ************* BXE
aws_secret_access_key=bheyut ******************* ywtRuUfChDL5r

Start the development server with

ebs-django-docker-tutorial git:(master) docker-compose -f docker-compose.dev.yml up --build

Update your ALLOWED_HOSTS in your settings.py file to include 127.0.0.1 . Now head over to postman and test the development server with an empty POST request at http://127.0.0.1/inventory-item .

That's it.

Conclusion

You now have a fair understanding of Elastic Beanstalk. In this article you have learned:

What Elastic Beanstalk is
How to integrate Nginx as a reverse proxy for your web server
How to deploy your Django app into a beanstalk docker environment
How to integrate an RDS instance into your EB environment.
How to mimic your remote EB environment into your development environment when using docker-compose.

If you found this article useful or learned something new, consider leaving a thumbs up and following me to keep up-to-date with any recent postings!

Till next time, happy coding!

Levi

Understanding Amazon SQS with Python and Django - Part 2

Ademola Akinsola — Sat, 13 Aug 2022 17:58:36 +0000

Hello👋, This is part 2 of my two-part series on Understanding Amazon SQS with Python and Django. This article assumes you have read the first article in the series; You can find that article at Understanding Amazon SQS with Python and Django - Part 1. The corresponding code can be found at this github repo.

Objectives

Learn how to consume messages from Amazon SQS with Python.
Create an endpoint to keep track of our File Processing.

How the file processing service (Consumer) interacts with the queue

There are a few things we should know about interacting with a queue when polling messages.

We can have several consumers polling messages from a single queue at once.
A Consumer must delete a message immediately after processing, to avoid another consumer from picking the same message up.
There is something called a VisibilityTimeout. This is the amount of time a consumer has to process and delete a message it has polled, to prevent other consumers from picking that same message up. When a message is polled by a consumer, that message is hidden from other consumers for the duration set in the VisibilityTimeout option. Default is 30 seconds.
You can learn more about the possibilities SQS provides us when polling messages in the Official Docs

Creating service B - The file processing service

Our service B is going to be a simple python script that runs infinitely and continually polls messages from our Queue to be processed. Let's create a new file process_messages_from_queue.py; Since this file is not actually part of our "Django app", we will create it outside our Django app directory to simulate it being on a completely different server. Your File tree should now look something like:

.
 amazon-sqs-and-django
    core
    db.sqlite3
    file
    manage.py
    test_file.txt
    requirements.txt
    venv
 process_messages_from_queue.py

First, let's create a separate environment for our script and install everything it would need. In your terminal, run the following:

amazon-sqs-and-django git:(part-2) python3 -m venv serviceb_venv             
amazon-sqs-and-django git:(part-2) source serviceb_venv/bin/activate
(serviceb_venv) amazon-sqs-and-django git:(part-2) pip install boto3

Now, inside the process_messages_from_queue.py file,

import json
import sqlite3
import boto3

session = boto3.Session(
    aws_access_key_id='<AWS_ACCESS_KEY_ID>', # replace with your key
    aws_secret_access_key='<AWS_SECRET_ACCESS_KEY>', # replace with your key
)
sqs = session.resource('sqs', region_name='us-east-1')
connection = sqlite3.connect("path/to/django/app/db.sqlite3")

# Statuses we defined in the File Model of our django app.
PROCESSING_STATUS = 1
PROCESSED_STATUS = 2
FAILED_STATUS = 3

def main():

    # Get the SQS queue we created.
    queue = sqs.get_queue_by_name(QueueName="MyFileProcessingQueue.fifo")

    # This loop runs infinitely since we want to constantly keep checking
    # if new messages have been sent to the queue, and if they have, retrieve them and process them.
    while True:
        cursor = connection.cursor()
        # retrieve some messages from the queue.
        messages = queue.receive_messages()

        for message in messages:
            data = json.loads(message.body)
            file_id = data["file_id"]

            # Update File to indicate that it is now in processing stage.
            cursor.execute("UPDATE file_file SET status = ? WHERE id = ?", (PROCESSING_STATUS, file_id,))
            connection.commit()

We create an sqs resource to interact with our queue. We also create a connection to our django app database; This is because we need to update the DB after processing the file. Then, we create a main function that will hold all of the important code. Inside the main function, we:

Retrieve the SQS queue and inside an infinite loop, continuously poll messages in the queue.
For each message, we get the file_id which was sent from the django app.
We use the file_id to update the File object in the database, and set its status to PROCESSING. This is to let the user know his file is no longer PENDING but rather, that work has started on it.

Things to note:

Right now, the receive_messages method only retrieves one message at a time, this may help to prevent overloading the server with too many requests at once. You can increase that value to a maximum of 10 by setting the MaxNumberOfMessages argument.

You can and should optimize the code to update the default of VisibilityTimeout if the service is unable to process a particular file fast enough. This is to prevent another queue from picking it up before it is done. You can do this by calling the change_visibility method on a particular message.

Now, let's actually process the file, and update the database with that processed data. And if the processing fails, we update the database to reflect the failure.

import os # <----- NEW
...
def main():
    ...
    while True:
        ...
        for message in messages:
            ...
            # <------- ADD THIS -------->
            # Get the File obj from the database through the file_id we got from the SQS message.
            file_object = cursor.execute("SELECT lines_count, file_size, file_path, status \
                FROM file_file WHERE id = ?", (file_id,)).fetchone()

            # file_object order ------ (lines_count, file_size, file_path, status)
            file_path = file_object[2]

            # Checking the state of the File Object before we start processing...
            print(f"FILE ID: {file_id}, LINES_COUNT: {file_object[0]}, FILE_SIZE in bytes: {file_object[1]}, STATUS: {file_object[3]}")

            try:
                lines = None
                with open(file_path, "r", encoding="utf-8") as file:
                    lines = len(file.readlines())
                file_size = os.path.getsize(file_path)

                cursor.execute("UPDATE file_file SET status = ?, lines_count = ?, file_size = ? WHERE id = ?",
                    (PROCESSED_STATUS, lines, file_size, file_id,))
            except Exception:
                cursor.execute("""UPDATE file_file SET status = ? WHERE id = ?""",
                    (FAILED_STATUS, file_id,),
                )

            connection.commit()
            # Delete the message, to avoid duplicate processing
            message.delete()

            # check updated database
            updated_file_object = cursor.execute("SELECT lines_count, file_size, file_path, status \
                FROM file_file WHERE id = ?", (file_id,)).fetchone()
            print(f"FILE ID: {file_id}, LINES_COUNT: {updated_file_object[0]}, FILE_SIZE in bytes: {updated_file_object[1]}, STATUS: {updated_file_object[3]}")
            # <------- END-------->

A quick brief on what is going on here:

We retrieve the file_path from the database, and get the number of lines in the file, as well as its size.
We save this processed data inside the database.
In the case an error occurs while processing, we update the database object to a FAILED_STATUS.
Lastly, we delete the message, so no other consumer can pick up the message again.

Note: The file processing action could be a bit more complex. Usually, it would be working on much larger files that take a lot longer to process, but for example purposes, we keep it simple.

Now, all that is left is to call our main function when our script is run.

import os
import json
import sqlite3
import boto3
...
def main():
    ...

if __name__ == " __main__":
    main()

The full code would now be:

import os
import json
import sqlite3
import boto3

session = boto3.Session(
    aws_access_key_id='<AWS_ACCESS_KEY_ID>', # replace with your key
    aws_secret_access_key='<AWS_SECRET_ACCESS_KEY>', # replace with your key
)
sqs = session.resource('sqs', region_name='us-east-1')
connection = sqlite3.connect("path/to/django/app/db.sqlite3")

# Statuses we defined in the File Model of our django app.
PROCESSING_STATUS = 1
PROCESSED_STATUS = 2
FAILED_STATUS = 3

def main():

    # Get the SQS queue we created.
    queue = sqs.get_queue_by_name(QueueName="MyFileProcessingQueue.fifo")

    # This loop runs infinitely since we want to constantly keep checking
    # if new messages have been sent to the queue, and if they have, retrieve them and process them.
    while True:
        cursor = connection.cursor()
        # retrieve some messages from the queue.
        messages = queue.receive_messages()

        for message in messages:
            data = json.loads(message.body)
            file_id = data["file_id"]

            # Update File to indicate that it is now in processing stage.
            cursor.execute("UPDATE file_file SET status = ? WHERE id = ?", (PROCESSING_STATUS, file_id,))
            connection.commit()

            # Get the File obj from the database through the file_id we got from the SQS message.
            file_object = cursor.execute("SELECT lines_count, file_size, file_path, status \
                FROM file_file WHERE id = ?", (file_id,)).fetchone()

            # file_object order ------ (lines_count, file_size, file_path, status)
            file_path = file_object[2]

            # Checking the state of the File Object before we start processing...
            print(f"FILE ID: {file_id}, LINES_COUNT: {file_object[0]}, FILE_SIZE in bytes: {file_object[1]}, STATUS: {file_object[3]}")

            try:
                lines = None
                with open(file_path, "r", encoding="utf-8") as file:
                    lines = len(file.readlines())
                file_size = os.path.getsize(file_path)

                cursor.execute("UPDATE file_file SET status = ?, lines_count = ?, file_size = ? WHERE id = ?",
                    (PROCESSED_STATUS, lines, file_size, file_id,))
            except Exception:
                cursor.execute("""UPDATE file_file SET status = ? WHERE id = ?""",
                    (FAILED_STATUS, file_id,),
                )

            connection.commit()

            # Delete the message, to avoid duplicate processing
            message.delete()

            # check updated database
            updated_file_object = cursor.execute("SELECT lines_count, file_size, file_path, status \
                FROM file_file WHERE id = ?", (file_id,)).fetchone()
            print(f"FILE ID: {file_id}, LINES_COUNT: {updated_file_object[0]}, FILE_SIZE in bytes: {updated_file_object[1]}, STATUS: {updated_file_object[3]}")

if __name__ == " __main__":
    main()

Testing the file processing service

Inside your terminal run the process_messages_from_queue.py script.

(serviceb_venv) amazon-sqs-and-django git:(part-2) python3 process_messages_from_queue.py

Now, open a second terminal, and start up the django app server. Remember, you would need to activate the virtual environment specific to the django app in the second terminal. Now, Let's go to postman and send a new POST request to process our test_file.txt file. What you should see, eventually, in your terminal is something like:

Great! Through the print statements, we can see that the File obj, with ID 8 in my case, which was initially empty and in PROCESSING or 1 state eventually got populated with the processed data and updated to PROCESSED or 2 state. This is what SQS can offer us; it allows us to postpone the actual processing, and we update the database with our results when it is done.

One Producer, Multiple Consumers

At this point, we already know SQS limits the rate at which requests are processed which in turn helps to avoid overloading a server. But, If we needed to process more messages at a quicker rate, we could add more consumer instances to split the load, reducing the total processing time required. It would look something like this:

Creating an endpoint to check file processing result/status

We received the message, processed the file, and saved the data, We still need to be able to let the user view the result of the process; if it failed or not. Update the FileView in the views.py file of our Django app like so:

...
from rest_framework.response import Response
from .models import File
from django.shortcuts import get_object_or_404 # <----- NEW

Class FileView(APIView):
    def post(self, request):
        ....

    # <----- ADD THIS ----->
    def get(self, request, pk: int):
        file = get_object_or_404(File, pk=pk)
        return Response({
                'lines_count': file.lines_count,
                'file_size': file.file_size,
                'status': file.status
            })
    # <-------- END -------->

Next up, update our project urls.py file.

from django.urls import path
from .views import FileView

urlpatterns = [
    path('process', FileView.as_view()),
    path('check-status/<int:pk>', FileView.as_view()), # <---- NEW
]

Now, inside postman, we can query our new check-status endpoint. We will replace pk with our file object id, in my case, that would be 8.

Now, a user can always confirm the status of their file processing request, which could be PENDING, PROCESSING, FAILED, or PROCESSED (In which case, the processed data is also shown).

Conclusion

And there you have it! You now have a fair understanding of SQS and how to integrate it into an application. Thanks for reading and coding along in this 2 part series on Understanding Amazon SQS with Python and Django. If you missed it, heres part 1 Understanding Amazon SQS with Python and Django - Part 1.

If you found this article useful or learned something new, consider leaving a thumbs up and following me to keep up-to-date with any recent postings!

Till next time, happy coding!

Levi

Understanding Amazon SQS with Python and Django - Part 1

Ademola Akinsola — Sun, 24 Jul 2022 15:18:00 +0000

Objectives

Introduce Amazon SQS and Queues.
Look into the different types of queues and their differences.
Understand how Amazon SQS would be used in a decoupled application or service.
Build a simple Django app integrating Amazon SQS.

Prerequisites

Good knowledge of python.
Knowledge of Django Rest Framework.

What is Amazon SQS?

Amazon Simple Queue Service (SQS) is a fully managed message queuing service that enables you to decouple and scale microservices, distributed systems, and serverless applications. Simply put, It is a service provided by AWS that uses the queue data structure underneath, to store messages it receives until they are ready to be retrieved.

What exactly is a queue?

A Queue is a data structure that takes in and returns data in the order of F.I.F.O (First in First Out). It's the same as a real-world queue, the first person to enter the queue is the first one to leave as well.

SQS Queues

SQS queues are queues we send messages to (JSON, XML, e.t.c.) and then later poll to retrieve those messages. The service sending the message is called a Producer and the service polling the message is called a Consumer. The size limit of a single message is 256KB.

Image from dzone.com

Amazon SQS provides us with two types of message queues:

Standard Queue : These queues provide at-least-once delivery of messages, this means all messages are delivered at least once , however, some messages may be delivered more than once. It also provides best-effort ordering , which means its possible messages are not delivered in the order that they were received.
FIFO Queue (First-in-First-out): These are queues designed for Only-Once delivery , meaning all messages will be processed exactly once. FIFO queue also guarantees ordered delivery , i.e, The FIRST message IN, is the FIRST message OUT(FIFO).

Looking at the above explanations, we might want to ask ourselves, Why exactly would we want to use a Standard Queue? Well,

Standard Queues allow near an unlimited number of transactions per second while FIFO queues only allow processing of a max of 300 messages transactions per second without batching, and 3000 with batching.
Standard queue is currently supported in all AWS regions, while FIFO, at the time of writing, is not.

Both are suited for different use cases. A FIFO queue is well suited for applications that highly depend on the order of the messages and have a low tolerance for duplicated messages. A standard Queue is suited for applications that are able to process duplicate and out-of-order messages.

Advantages of Amazon SQS

Increased performance - SQS allows for asynchronous communication between different parts of an application.
Increases Reliability - If the consumer service fails or crashes. Since the messages still exist in the queue, it can pick them up again when it's back online. This leads to increased reliability.
Scalability - A 'standard SQS' queue allows close to an unlimited number of message transactions per second. This makes it easier for your service to scale from thousands of requests per second to millions.
Buffer requests - SQS helps to prevent overloading a service with crazy amounts of requests. With SQS, the service can choose how many requests it wants to work on at any given time, and SQS will never return more than this to the service.

Example use-case for Amazon SQS

Imagine you have a service that currently:

Takes in URL paths to large files from a user.
Does some time-consuming processing on the file, and saves the result into the database.
And eventually returns the result to the user.

What are the possible problems here?

If service B gets overloaded with a high amount of files to process, it could slow the server down, eventually leading to the server crashing.
If the entire server ever happens to go down, it would lead to failure for all users currently using the service to process files.

How can amazon SQS be used to solve this? Using SQS, we can take out our file processing function and place it on a new dedicated server, hence "decoupling" service A from the file processing function(service B).

The flow of the service using SQS could now be as follows:

The user sends a request to our server 1 with the file path for processing.
Service A saves the name of the file and the file path to the database with a status field of "Pending" and gets the ID of the newly saved object.
Service A sends a message to the queue containing the ID of the saved object before returning a response to the user indicating the file is now in the queue pending processing.
Service B polls specific amounts of messages from the queue to process at its own pace(This prevents the server from getting overloaded and crashing). It then gets the ID's from those messages.
Service B uses the ID to retrieve the saved object from the database, get the file path, and start processing the file.
Once it succeeds or fails, it updates the stored object to an appropriate status (Succeeded/Failed).

Server 1 also has an endpoint that users can utilize to check up on the status of their file processing request and its data if it has already been processed successfully.

Project Setup

We will be building out the file processing service example. For this, we would need to create an aws account and get our aws access and secret access keys from the console. Once you've gotten your keys store them safely.

Let's also create our first Queue. With the search bar, navigate to the Amazon SQS.

Now on the SQS page, click on Create queue. We are taken to the SQS Queue creation page and we see some options presented to us.

For this tutorial, Choose the FIFO Queue, and then enter a name for your queue. Note that FIFO queue names must end with .fifo. Look for the Content-based deduplication option and turn it on(This helps to prevent consumers from picking up a possible duplicate message from the queue). Leave every other option as the default and click Create queue at the bottom. Copy the name of the Queue, in my case, MyFileProcessingQueue.fifo, and store it somewhere, we would be using it later. Also, take note of the region name where your queue is created, you can view it from the top right of the console. For me, that would be us-east-1.

Now into the Django part, let's begin by creating a new directory and setting up a new project.

$ mkdir amazon-sqs-django && cd amazon-sqs-django
$ python3.8 -m venv venv # --> Create virtual environment.
$ source venv/bin/activate # --> Activate virtual environment.

(venv)$ pip install django==4.0.5 djangorestframework==3.13.1
(venv)$ django-admin startproject core .

We will also install boto3 to help us communicate more easily with our SQS Queue.

(venv)$ pip install boto3==1.24.20

Next, let's create a Django app called file which will hold the models and logic for the entire service A.

(venv)$ python manage.py startapp file

# *core/settings.py*

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'file', # new
]

Creating our File model

Inside file/models.py, add the following

# *file/models.py*

...
class File(models.Model):
    class FileStatus(models.IntegerChoices):
        PENDING = 0
        PROCESSING = 1
        PROCESSED = 2
        FAILED = 3

    lines_count = models.IntegerField(null=True)
    file_size = models.IntegerField(null=True)
    file_path = models.CharField(max_length=120)
    status = models.IntegerField(choices=FileStatus.choices, default=FileStatus.PENDING)
...

This model will store some basic information that we "process" from the files. The status indicates what stage of the processing, the actual file is in. We have four status types:

PENDING : This means the file has not yet been picked up for processing.
PROCESSING : The file has been picked up, and work is being done on it currently.
PROCESSED : The file was successfully processed; no issues were encountered.
FAILED : Something went wrong somewhere; the file could not be processed correctly.

Creating the view

Inside the views.py, let's add the following

import boto3

from rest_framework.views import APIView
from .models import File

session = boto3.Session(
    aws_access_key_id='<AWS_ACCESS_KEY_ID>', # replace with your key
    aws_secret_access_key='<AWS_SECRET_ACCESS_KEY>', # replace with your key
)
sqs = session.resource('sqs', region_name='<AWS_REGION_NAME>') # replace with your region name

class FileView(APIView):
    """
    Process file and saves its data.
    :param file_path: path to file on a remote or local server
    :return: status
    """
    def post(self, request):
        file_path = request.data.get('file_path')
        file_obj = File.objects.create(file_path=file_path) # save file unprocessed.

        # Get our recently created queue.
        queue = sqs.get_queue_by_name(QueueName="MyFileProcessingQueue.fifo")

Using boto3, we instantiate a utility class sqs for interacting with the Amazon SQS resource. Then we create a view and now inside its post method, we get the file path from the user and save it into the database. The saved File object has a default status of PENDING which we previously set in the model class. We also get our queue by its name, which in my case, is MyFileProcessingQueue.fifo.

To send a message to our queue, let's update the view

import json # new
import boto3

from rest_framework.views import APIView
from rest_framework.response import Response # new
from .models import File

.....

class FileView(APIView):

    def post(self, request):
        .....
        message_body = {
            'file_id': str(file_obj.id)
        }

        # Send a message to the queue, so we can process this particular file eventually.
        response = queue.send_message(
            MessageBody=json.dumps(message_body),
            MessageGroupId='messageGroupId'
        )

        # Let the user know the file has been sent to the queue and is PENDING processing.
        return Response({"message": "File has been scheduled for processing..."}, 
    status=200)

Here,

We send an sqs message containing our saved File object ID, so we can later use it to retrieve the object from the database and get the file path to be processed. TheMessageGroupId ensures that all messages with the same MessageGroupId are processed in a FIFO order. Usually, this would be set to something unique like the user ID or session ID, but for simplicity, we use the string 'messageGroupId'. You can learn more about the possible parameters for the send_message method.
After sending the message to our SQS queue. We send a response immediately to the user letting them know their file is now in the queue pending processing, so they can go about their activities with our service without concern.

Your view should now look something like this:

# *file/views.py*
import json
import boto3

from rest_framework.views import APIView
from rest_framework.response import Response
from .models import File

session = boto3.Session(
    aws_access_key_id='<AWS_ACCESS_KEY_ID>', # replace with your key
    aws_secret_access_key='<AWS_SECRET_ACCESS_KEY>', # replace with your key
)
sqs = session.resource('sqs', region_name='<AWS_REGION_NAME>') # replace with your region name

class FileView(APIView):
    """
    Process file and saves its data.
    :param file_path: path to file.
    :return: status
    """
    def post(self, request):
        file_path = request.data.get('file_path')
        file_obj = File.objects.create(file_path=file_path) # save file unprocessed.

        # Get our recently created queue.
        queue = sqs.get_queue_by_name(QueueName="MyFileProcessingQueue.fifo")
        message_body = {
            'file_id': str(file_obj.id)
        }

        # Send a message to the queue, so we can process this particular file eventually.
        response = queue.send_message(
            MessageBody=json.dumps(message_body),
            MessageGroupId='messageGroupId'
        )

        # Let the user know the file has been sent to the queue and is PENDING processing.
        return Response({"message": "File has been scheduled for processing..."}, 
    status=200)

Testing what we have so far

First, let's connect the file app URLs to our project URLs. Inside the file folder, create a new file urls.py.

# file/urls.py

from django.urls import path
from .views import FileView

urlpatterns = [
    path('process', FileView.as_view())
]

Now edit the core/urls.py file,

from django.contrib import admin
from django.urls import path, include # new

urlpatterns = [
    path('admin/', admin.site.urls),
    path('files/', include('file.urls')), # new
]

Next, make and run migrations

(venv)$ python manage.py makemigrations
(venv)$ python manage.py migrate

Now start the development server.

(venv)$ python manage.py runserver

For easy testing, we will create a file inside the root folder named test_file.txt and fill it up with 10 or more lines just so it's not empty. This will be our file to be processed. Now your folder path should look like

amazon-sqs-and-django
 core
 db.sqlite3
 file
 manage.py
 test_file.txt
 venv

Now, launch postman and test the http://127.0.0.1:8000/files/process endpoint with the absolute file path to the test_file.txt.

Now, let's head back over to the queues page on the AWS console and click our recently created queue - MyFileProcessingQueue.fifo. Then, on the top right of the new page, click Send and Receive messages.

Now, click on poll messages and click the first message that comes in, you should see the data we sent from the Django app.

Now we know the first part of our service works. We are now able to send a message from our Django app into an SQS queue.

Conclusion and Next Steps

In this first part of this AWS SQS python and Django article, you learned the basics of queues in general and Amazon SQS, its advantages, and some of its use cases. We learned how to create a queue and built a Django app to send messages to that queue.

In the next article, we will

Build out our service B, the File processing service that will consume messages from the queue, process the file, and update the database.
We will also provide an endpoint for the user to check on the result of the file processing at any given point in time.

If you found this article useful or learned something new, consider leaving a thumbs up!

Till next time, happy coding!

Levi