Complete RESTAPI solution with Flask

In this article I will mention how to develop an APIREST in Flask with migration to the postgregis2 database and migrating with docker. This was a technical test for a company as a Python backend developer using the Flask framework.

Understanding the business rule

The technical test is aimed at understanding the developer’s development practices and thus creating an application where we can read, create, update and delete and thus carry out authentication for each route and develop a complete application. Now that we understand the business rules, let’s analyze and start coding the project completely.

Let’s start coding

I will mention each concept topic by topic and how to choose each concept to solve this problem for the company.

My review:

As an introduction to analysis, we will choose Flask for web development framework which is aimed at the core Python, SQLAlchemy used for ORM and Docker for generate the PostGIS image to work with geospatial data.

Project Environment Settings

Configuring the PostgreSQL Environment with Docker 🔧 🐳

Certainly! Let’s create a step-by-step guide to create and run a PostgreSQL container using Docker and Docker Compose.

Create the Dockerfile.

FROM python:3.11

RUN apt-get update \
    && apt-get install -y postgresql postgis \
    && apt-get clean \
    && rm -rf /var/lib/apt/lists/*

ENV POSTGRES_DB=apirestflask
ENV POSTGRES_USER=root
ENV POSTGRES_PASSWORD=1234

RUN mkdir -p /etc/postgresql/12/main/ \
    && echo "host    all    all    0.0.0.0/0    md5" >> /etc/postgresql/12/main/pg_hba.conf

RUN service postgresql start \
    && su postgres -c "psql -U postgres -c 'CREATE DATABASE apirestflask;'"

WORKDIR /app

COPY . /app

RUN apt-get update \
    && apt-get install -y python3-dev libpq-dev python3-pip \
    && pip install --no-cache-dir -r requirements.txt

EXPOSE 5000

CMD ["python", "run.py"]

Create a file called docker-compose.yml

# docker-compose.yml
version: '3'
services:
  postgres:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "5432:5432"
    volumes:
      - ./database:/var/lib/postgresql/data
    environment:
      POSTGRES_PASSWORD: 1234
      POSTGRES_DB: apirestflask
      POSTGRES_USER: root

This docker-compose.yml file defines the PostgreSQL service, specifying the build from the Dockerfile, the port mapped to the host, volumes for data persistence, and the environment variables.

Build the image and start the container

In the terminal, run the following commands:

docker-compose build
docker-compose up -d
docker-compose ps -a 

Connect to PostgreSQL

You can use a database administration tool such as psql or a graphical tool such as pgAdmin to connect to PostgreSQL. Use the following settings:

• Host: localhost
• Port: 5432
• Database name: flaskrestapi
• Username: root
• Password: 1234

Stop and remove container (Optional).

When you are finished using the container, stop and remove it with the following command:

docker-compose down

Project configuration in Python: 🐍🔎💻

Python configuration the project was developed with package pypi

But quote and show how you install the dependencies with this package installer.

Creating your virtual environment:

python -m venv env # or venv 

According to the command it will create a venv or env of your choice within the folder you want to clone this project.

Cloning this project:

git clone https://github.com/Hedriss10/restapi_flask_test_tecnico.git

Cloning with web URL but you can also clone via ssh or GitHub CLI depending on your preference.

Installing the dependencies:

pip install -r requirements.txt

According to requiremenets.txt it is responsible for maintaining the frameworks in our application.

How to execute the project 🖥️🌐

To run the api server, open the terminal in the directory where you cloned the project and type the following commands:

$env:FLASK_ENV = "development"
$env:FLASK_APP = "run"
python run.py # debuggers 

Now let’s understand the business rule with ORM, our models and resources

In this code I will show all the API models with CRUD and user registration

from geoalchemy2 import Geometry
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import func

db = SQLAlchemy()

class ItemModel(db.Model):
    """
    In this ORM model with SQLAlchemy and geoalchemy, we will create the database with the following schema:
    
    We define the table name as `items`.
    
    id: We define a column named `id` that stores up to 30 characters and contains a primary key.
    
    name: We define a column named `name` that stores up to 30 characters and cannot be a null field.
    
    description: We define a column named `description` that stores text and cannot be null.
    
    geometry: We define a column named `geometry` that stores geometric data of type (point) with spatial reference system SRID(4326) and also does not allow null values.
    """
    __tablename__ = "items"
    
    id = db.Column(db.String(30), primary_key=True)
    name = db.Column(db.String(30), nullable=False)
    description = db.Column(db.Text, nullable=False)
    geometry = db.Column(Geometry('POINT', srid=4326), nullable=False)
    
    # Class initializer for the Item
    def __init__(self, id, name, description, geometry):
        self.id =  id 
        self.name = name
        self.description = description 
        self.geometry = geometry
        
    # Function to return a JSON representation
    def json(self):
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "geometry": str(self.geometry)  # Converting geometry to WKT
        }
        
    # Function to find an item by ID     
    @classmethod
    def find_id(cls, id):
        item = cls.query.filter_by(id=id).first()
        return item
    
    # Function to save the item to the database
    def save_id(self):
        db.session.add(self)
        db.session.commit()
        
    # Function to update the item 
    def update_id(self, id, name, description, geometry):
        self.id = id 
        self.name =  name 
        self.description = description
        self.geometry = geometry
        
    # Function to delete from the database 
    def delete_id(self):
        db.session.delete(self)
        db.session.commit()
    
    # Function to find items within a specific radius 
    @classmethod
    def find_nearest_satellite(cls, latitude, longitude, radius):
        """
        First, declare a variable 'point' and use PostgreSQL's ST_MakePoint with the provided latitude and longitude coordinates. 
        The ST_SetSRID function is used to explicitly set the SRID (Spatial Reference System) as 4326, ensuring it matches the SRID of the geometry in the 'cls.geometry' column in the database.

        ---
        Second, declare a variable 'result' is used to apply the ST_DWithin condition, which checks if the geometry of the table ('cls.geometry') is within the specified radius of the point created earlier. 
        Then, 'order_by' is used to sort the results by the distance between the geometry and the point. 
        Finally, 'all()' is called to execute the query and get a list of results.
        """
        point = func.ST_SetSRID(func.ST_MakePoint(longitude, latitude), 4326)

        result = cls.query.filter(
            func.ST_DWithin(cls.geometry, point, radius)
        ).order_by(
            func.ST_Distance(cls.geometry, point)
        ).all()  # Using the 'all()' method to get a simple list 

        return result

class UsersModel(db.Model):
    """
    In this class, we will create the user registration.
    We define a column named 'user_id' of type integer and it is a primary key.
    We define a column named 'login' that contains a field of up to 40 characters.
    We define a column named 'password' that contains a field of up to 40 characters.
    """
    __tablename__ = "users"
    
    user_id = db.Column(db.Integer, primary_key=True)
    login = db.Column(db.String(40))
    password = db.Column(db.String(40))
    
    def __init__(self, login, password):
        self.login = login
        self.password = password

    def json(self):
        return {
            'user_id': self.user_id,
            'login': self.login
        }

    # Function to find a user by ID         
    @classmethod
    def find_user(cls, user_id):
        user = cls.query.filter_by(user_id=user_id).first()
        if user:
            return user
        return None

    # Function to find a user by login 
    @classmethod
    def find_by_login(cls, login):
        user = cls.query.filter_by(login=login).first()
        if user:
            return user
        return None

    # Function to save the user
    def save_user(self):
        db.session.add(self)
        db.session.commit()

    # Function to delete the user
    def delete_user(self):
        db.session.delete(self)
        db.session.commit()

See the resource code and system applicability

from flask_restx import Resource, fields, Api, reqparse
from flask_jwt_extended import jwt_required, create_access_token
from models.GeosCrud import ItemModel, UsersModel

api = Api()

item_model = api.model('Item', {
    'id': fields.String(required=True, description='Unique identifier of the item'),
    'name': fields.String(required=True, description='Name of the item'),
    'description': fields.String(required=True, description='Description of the item'),
    'geometry': fields.String(required=True, description='Geometry field of type Point')
})

geospatial_search_model = api.model('GeospatialSearch', {
    'latitude': fields.Float(required=True, description='Latitude of the central point'),
    'longitude': fields.Float(required=True, description='Longitude of the central point'),
    'radius': fields.Float(required=True, description='Search radius in specific units')
})

class ItemAllResource(Resource):
    """
    This class returns all items registered in the database.
    """
    def get(self):
        return {"message": [item.json() for item in ItemModel.query.all()]}

class ItemResource(Resource):
    attrs = reqparse.RequestParser()
    attrs.add_argument("id", type=str)
    attrs.add_argument("name")
    attrs.add_argument("description")
    attrs.add_argument("geometry")

    # Implementation to get an item by ID
    @jwt_required()
    @api.expect(id)
    def get(self, id):
        search_id = ItemModel.find_id(id=id)
        if search_id:
            # Using the json method for serialization
            return search_id.json(), 200, {'Content-Type': 'application/json'}
        return {"message": "Id not found."}, 404

    # Implementation to create a new item
    @jwt_required()
    @api.expect(id)
    def post(self, id):
        if ItemModel.find_id(id=id):
            return {"message": "Item id '{}' already exists".format(id)}

        data = ItemResource.attrs.parse_args()
        new_item = ItemModel(id=data['id'], name=data['name'], description=data['description'], geometry=data['geometry'])

        try:
            new_item.save_id()
        except Exception as e:
            print(e)
            return {"message": "An error occurred trying to create item"}, 500

        return new_item.json(), 201

    # Implementation to update an item by ID
    @jwt_required()
    @api.expect(id)
    def put(self, id):
        data =  ItemResource.attrs.parse_args()
        search_id = ItemModel(id=data['id'], name=data['name'], description=data['description'], geometry=data['geometry'])

        new_search_id = ItemModel.find_id(id=id)
        if new_search_id:
            new_search_id.update_id(**data)
            new_search_id.save_id()
            return new_search_id.json(), 200

        search_id.save_id()
        return search_id.json(), 201

    # Implementation to delete an item by ID
    @jwt_required()
    @api.expect(id)
    def delete(self, id):
        id =  ItemModel.find_id(id=id)
        if id:
            id.delete_id()
            return {"message" : "Item deleted."}
        return {"message": "Item not found."}, 404

class GeospatialSearch(Resource):
    """
    Class for performing the search for an item by a specific point
    Requiring user authentication
    """
    @jwt_required()
    # @api.expect(api.model('GeospatialSearch', {
    #     'latitude': fields.Float(required=True, description='Latitude of the central point'),
    #     'longitude': fields.Float(required=True, description='Longitude of the central point'),
    #     'radius': fields.Float(required=True, description='Search radius in specific units')
    # }))
    def get(self):
        """
        URI get method to search for the item
        Validating the presence and correct format of the parameters
        Performing the geospatial search with GeoAlchemy
        Returning via JSON
        """
        args = api.payload
        latitude = args['latitude']
        longitude = args['longitude']
        radius = args['radius']

        if not (latitude and longitude and radius):
            return {"message": "Latitude, longitude, and radius are required parameters."}, 400

        result = ItemModel.find_nearest_satellite(latitude, longitude, radius)

        return {"results": [item.json() for item in result]}, 200

class UserRegistrationResource(Resource):
    """
    Class to register the user in the database, which involves receiving arguments like login and password
    According to the development, SQLAlchemy already creates the id, it's worth noting to continue with this API model
    """
    attrs = reqparse.RequestParser()
    attrs.add_argument('login', type=str)
    attrs.add_argument('password', type=str)

    # Implementation to register a new user
    def post(self):
        data = UserRegistrationResource.attrs.parse_args()
        if UsersModel.find_by_login(data['login']):
            return {"message" : "The login '{}' already exists".format(data['login'])}

        user = UsersModel(**data)
        user.save_user()
        return {'message': 'User created successfully!'}, 201  # Created

class UserAuthenticationResource(Resource):
    """
    Class to create a jwt access token and pass it to the 'access' to the user
    Creating a URI post method
    """
    attrs = reqparse.RequestParser()
    attrs.add_argument('login', type=str)
    attrs.add_argument('password', type=str)

    @classmethod
    def post(cls):
        """
        We declare a variable 'data' where we take the attributes of the class and pass them with **kwargs
        then we perform a simple search with 'user' where we compare the login in the database and the login passed via json
        then we create the token with the create_access_token method from jwt and pass it with the status code 200, and finally
        we return an error message if the user is providing the wrong login and password
        """
        data = UserAuthenticationResource.attrs.parse_args()
        user = UsersModel.find_by_login(data['login'])

        if user and user.password == data['password']:
            access_token = create_access_token(identity=user.user_id)
            return {'access_token': access_token}, 200

        return {'message': 'The username or password is incorrect.'}, 401

Finally, our Flask app:

# --utf8--
"""
Flask app with business rules applied according to the technical test
"""
from flask import Flask
from flask_restx import Api
from flask_jwt_extended import JWTManager
from models.GeosCrud import db
from resources.geospatial import (ItemResource, GeospatialSearch, 
                                  UserRegistrationResource, UserAuthenticationResource, ItemAllResource)

from config import config_by_name

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config_by_name[config_name])
    
    db.init_app(app)  
    
    api = Api(app, version="1.0", title='APIREST Geospatial Items', description='Development of APIREST for technical test.')
    
    jwt = JWTManager(app)
    
    @app.before_request
    def before_first_request():
        with app.app_context():
            db.create_all()

    
    def load_endpoint_uri():
        api.add_resource(ItemAllResource, '/items')  # Retrieve all items from the database 
        api.add_resource(ItemResource, '/item/<string:id>')  # CRUD operations for items 
        api.add_resource(GeospatialSearch, '/geo-search')  # Search for an item by geometric field 
        api.add_resource(UserRegistrationResource, '/register')  # User registration
        api.add_resource(UserAuthenticationResource, '/login')  # User login
        
    load_endpoint_uri()
    
    return app

if __name__ == "__main__":
    app = create_app(config_name='development')
    app.run(host=app.config['IP_HOST'], port=app.config['PORT_HOST'], debug=app.config['DEBUG'])

I hope you enjoy this complete RESTAPI solution with Flask, analyze the code and see every detail and set up your project modularization. web developer community