Loading Now
WordPress

Create a REST API Using Flask on Ubuntu

Create a REST API Using Flask on Ubuntu

Flask, a lightweight web framework for Python, provides an efficient way for developers to create REST APIs on Ubuntu Servers. This guide will lead you through establishing production-ready Flask APIs, covering everything from initial setup to deployment tips. You will acquire crucial knowledge about API development patterns, database integration, authentication methods, and performance enhancement strategies that accommodate your application’s expansion.

<h2>Flask for REST API Development Explained</h2>
<p>Flask adheres to the WSGI standard, which makes it much lighter compared to Django’s feature-rich setup. Its straightforward nature is especially beneficial for API crafting, granting you complete control over endpoints, request management, and response formatting. In contrast to FastAPI’s automatic OpenAPI generation or the complexity of Django REST Framework’s serializers, Flask provides you with detailed oversight of every API aspect.</p>
<p>The framework’s handling of the request-response cycle utilizes decorators for HTTP methods, simplifying REST endpoint creation. Flask’s modular framework enables you to incorporate tools like SQLAlchemy for database tasks, Flask-JWT-Extended for authentication, or Flask-CORS for cross-origin resource sharing as needed.</p>

<h2>Setting Up Ubuntu Environment and Flask Installation</h2>
<p>To get Flask running on Ubuntu, ensure you have Python 3.8 or higher and manage virtual environments. Follow these steps for complete setup:</p>
<pre><code>sudo apt update && sudo apt upgrade -y

sudo apt install python3 python3-pip python3-venv
python3 -m venv flask_api_env
source flask_api_env/bin/activate
pip install –upgrade pip

Next, install Flask along with essential extensions for API development:

pip install flask flask-sqlalchemy flask-migrate flask-cors flask-jwt-extended python-dotenv

When it’s time for production, include the following packages:

pip install gunicorn psycopg2-binary redis

Create the structure for your project:

mkdir flask_rest_api && cd flask_rest_api
touch app.py config.py requirements.txt .env .gitignore
<h2>Crafting Your First REST API</h2>
<p>Begin with a straightforward Flask application structure aligned with REST principles:</p>
<pre><code>from flask import Flask, request, jsonify

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_cors import CORS
import os
from dotenv import load_dotenv

load_dotenv()

app = Flask(name)
app.config[‘SQLALCHEMY_DATABASE_URI’] = os.getenv(‘DATABASE_URL’, ‘sqlite:///api.db’)
app.config[‘SQLALCHEMY_TRACK_MODIFICATIONS’] = False
app.config[‘SECRET_KEY’] = os.getenv(‘SECRET_KEY’, ‘dev-key-change-in-production’)

db = SQLAlchemy(app)
migrate = Migrate(app, db)
CORS(app)

User model

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
created_at = db.Column(db.DateTime, default=db.func.current_timestamp())

def to_dict(self):
    return {
        'id': self.id,
        'username': self.username,
        'email': self.email,
        'created_at': self.created_at.isoformat()
    }

API Routes

@app.route(‘/api/users’, methods=[‘GET’])
def get_users():
users = User.query.all()
return jsonify([user.to_dict() for user in users])

@app.route(‘/api/users’, methods=[‘POST’])
def create_user():
data = request.get_json()

if not data or 'username' not in data or 'email' not in data:
    return jsonify({'error': 'Both username and email are required'}), 400

if User.query.filter_by(username=data['username']).first():
    return jsonify({'error': 'Username already taken'}), 409

user = User(username=data['username'], email=data['email'])
db.session.add(user)
db.session.commit()

return jsonify(user.to_dict()), 201

@app.route(‘/api/users/‘, methods=[‘GET’])
def get_user(user_id):
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())

@app.route(‘/api/users/‘, methods=[‘PUT’])
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()

if 'username' in data:
    user.username = data['username']
if 'email' in data:
    user.email = data['email']

db.session.commit()
return jsonify(user.to_dict())

@app.route(‘/api/users/‘, methods=[‘DELETE’])
def delete_user(user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return ”, 204

if name == ‘main‘:
with app.app_context():
db.create_all()
app.run(debug=True, host=”0.0.0.0″, port=5000

Initialize the database and activate your API:

export FLASK_APP=app.py
flask db init
flask db migrate -m "Initial migration"
flask db upgrade
python app.py

Verify your endpoints using curl:

curl -X POST http://localhost:5000/api/users \
      -H "Content-Type: application/json" \
      -d '{"username": "testuser", "email": "[email protected]"}'

curl http://localhost:5000/api/users

<h2>Enhanced API Features and Authentication</h2>
<p>For production-ready APIs, you need authentication, input validation, and error management. Here’s an upgraded version using JWT for authentication:</p>
<pre><code>from flask_jwt_extended import JWTManager, jwt_required, create_access_token, get_jwt_identity

from werkzeug.security import generate_password_hash, check_password_hash
from marshmallow import Schema, fields, ValidationError

app.config[‘JWT_SECRET_KEY’] = os.getenv(‘JWT_SECRET_KEY’, ‘jwt-secret-change-in-production’)
jwt = JWTManager(app)

Enhanced User model with password

class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, default=db.func.current_timestamp())

def set_password(self, password):
    self.password_hash = generate_password_hash(password)

def check_password(self, password):
    return check_password_hash(self.password_hash, password)

Validation schemas

class UserSchema(Schema):
username = fields.Str(required=True, validate=lambda x: len(x) >= 3)
email = fields.Email(required=True)
password = fields.Str(required=True, validate=lambda x: len(x) >= 6)

user_schema = UserSchema()

Authentication endpoints

@app.route(‘/api/auth/register’, methods=[‘POST’])
def register():
try:
data = user_schema.load(request.get_json())
except ValidationError as err:
return jsonify({‘errors’: err.messages}), 400

if User.query.filter_by(username=data['username']).first():
    return jsonify({'error': 'Username already exists'}), 409

user = User(username=data['username'], email=data['email'])
user.set_password(data['password'])
db.session.add(user)
db.session.commit()

access_token = create_access_token(identity=user.id)
return jsonify({'access_token': access_token, 'user': user.to_dict()}), 201

@app.route(‘/api/auth/login’, methods=[‘POST’])
def login():
data = request.get_json()
user = User.query.filter_by(username=data.get(‘username’)).first()

if user and user.check_password(data.get('password')):
    access_token = create_access_token(identity=user.id)
    return jsonify({'access_token': access_token})

return jsonify({'error': 'Invalid credentials'}), 401

Protected endpoint

@app.route(‘/api/profile’, methods=[‘GET’])
@jwt_required()
def get_profile():
user_id = get_jwt_identity()
user = User.query.get(user_id)
return jsonify(user.to_dict())

<h2>Integrating with Databases and Migration Practices</h2>
<p>Flask-SQLAlchemy offers excellent integration with PostgreSQL for production stages. Set up your PostgreSQL connection:</p>
<pre><code>sudo apt install postgresql postgresql-contrib

sudo -u postgres createuser –interactive
sudo -u postgres createdb flask_api_db

Revise your .env file:

DATABASE_URL=postgresql://username:password@localhost/flask_api_db
SECRET_KEY=your-secret-key-here
JWT_SECRET_KEY=your-jwt-secret-here

Manage database migrations effectively:

flask db migrate -m "Add password field to users"
flask db upgrade

For intricate queries and relationships:

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())

user = db.relationship('User', backref=db.backref('posts', lazy=True))


Add to User model


def to_dict_with_posts(self):

return {

**self.to_dict(),

'posts': [{'id': p.id, 'title': p.title} for p in self.posts]

}


<h2>Optimisation Techniques and Caching</h2>
<p>Use Redis for caching frequently accessed data:</p>
<pre><code>import redis


from functools import wraps

import json


redis_client = redis.Redis(host=”localhost”, port=6379, db=0, decode_responses=True)


def cache_result(timeout=300):

def decorator(f):

@wraps(f)

def decorated_function(*args, **kwargs):

cache_key = f”{f.name}:{hash(str(args) + str(kwargs))}”

cached_result = redis_client.get(cache_key)


        if cached_result:
            return json.loads(cached_result)

        result = f(*args, **kwargs)
        redis_client.setex(cache_key, timeout, json.dumps(result))
        return result
    return decorated_function
return decorator


@app.route(‘/api/users’, methods=[‘GET’])

@cache_result(timeout=600)

def get_users_cached():

users = User.query.all()

return [user.to_dict() for user in users]


<p>Incorporate database query optimisation:</p>
<pre><code># Pagination for large datasets


@app.route(‘/api/users’, methods=[‘GET’])

def get_users_paginated():

page = request.args.get(‘page’, 1, type=int)

per_page = min(request.args.get(‘per_page’, 20, type=int), 100)


users = User.query.paginate(
    page=page, per_page=per_page, error_out=False
)

return jsonify({
    'users': [user.to_dict() for user in users.items],
    'total': users.total,
    'pages': users.pages,
    'current_page': page
})</code></pre>

<h2>Comparison of Frameworks and Their Use Cases</h2>
<table>
    <thead>
        <tr>
            <th>Framework</th>
            <th>Complexity of Setup</th>
            <th>Performance</th>
            <th>Learning Curve</th>
            <th>Optimal For</th>
        </tr>
    </thead>
    <tbody>
        <tr>
            <td>Flask</td>
            <td>Low</td>
            <td>Good</td>
            <td>Gentle</td>
            <td>Small to medium APIs, microservices</td>
        </tr>
        <tr>
            <td>FastAPI</td>
            <td>Low</td>
            <td>Excellent</td>
            <td>Moderate</td>
            <td>High-performance APIs, async operations</td>
        </tr>
        <tr>
            <td>Django REST</td>
            <td>High</td>
            <td>Good</td>
            <td>Steep</td>
            <td>Complex applications, admin interfaces</td>
        </tr>
        <tr>
            <td>Express.js</td>
            <td>Low</td>
            <td>Excellent</td>
            <td>Moderate</td>
            <td>JavaScript ecosystem, real-time apps</td>
        </tr>
    </tbody>
</table>
<p>Flask is particularly adept in situations that require:</p>
<ul>
    <li>Quick prototyping and development</li>
    <li>Custom authentication frameworks</li>
    <li>Integration with pre-existing Python projects</li>
    <li>Microservices architecture</li>
    <li>Educational projects and learning about APIs</li>
</ul>

<h2>Deploying in Production with Gunicorn and Nginx</h2>
<p>Set up Gunicorn for serving your application in production:</p>
<pre><code>pip install gunicorn


gunicorn –bind 0.0.0.0:8000 –workers 4 –timeout 120 app:app


Create a systemd service file:


sudo nano /etc/systemd/system/flask-api.service

[Unit]
Description=Flask API
After=network.target


[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/your/flask_rest_api
Environment="PATH=/path/to/your/flask_rest_api/flask_api_env/bin"
ExecStart=/path/to/your/flask_rest_api/flask_api_env/bin/gunicorn --bind 127.0.0.1:8000 --workers 4 app:app
Restart=always


[Install]
WantedBy=multi-user.target




Configure Nginx to serve as a reverse proxy:


sudo apt install nginx
sudo nano /etc/nginx/sites-available/flask-api

server {
listen 80;
server_name your-domain.com;


location / {
    proxy_pass http://127.0.0.1:8000;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
}


}


sudo ln -s /etc/nginx/sites-available/flask-api /etc/nginx/sites-enabled/

sudo systemctl restart nginx


<h2>Frequent Mistakes and Solutions</h2>
<p>Common challenges in Flask development include:</p>
<p>**Database Connection Hiccups**: Always use connection pools and manage database errors properly:</p>
<pre><code>from sqlalchemy.exc import OperationalError


@app.errorhandler(OperationalError)

def handle_db_error(error):

db.session.rollback()

return jsonify({‘error’: ‘Could not connect to the database’}), 500


**CORS Complications**: Configure CORS correctly for seamless frontend interaction:


from flask_cors import CORS
CORS(app, origins=['http://localhost:3000', 'https://yourdomain.com'])


**Memory Leaks**: Ensure to close database connections and use connection pools:


@app.teardown_appcontext
def close_db(error):
    if hasattr(g, 'db_connection'):
        g.db_connection.close()


**Issues with JWT Tokens**: Develop proper token refresh capabilities and manage expired tokens:


@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
    return jsonify({'error': 'Token has expired'}), 401


Maintaining performance is vital as you scale. Incorporate logging and monitoring:


import logging
from flask.logging import default_handler

logging.basicConfig(level=logging.INFO)
app.logger.removeHandler(default_handler)


handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s'
))
app.logger.addHandler(handler)




<p>Flask’s versatility makes it a fantastic choice for API development when precise control is necessary. Its minimalist design enables you to introduce complexity only when required, which is beneficial for both grasping REST principles and creating full-fledged APIs. For detailed Flask documentation and advanced techniques, refer to the <a href="https://flask.palletsprojects.com/" rel="follow opener" target="_blank">official Flask documentation</a> and check out the <a href="https://flask-sqlalchemy.palletsprojects.com/" rel="follow opener" target="_blank">Flask-SQLAlchemy guide</a> for best practices in database integration.</p>
<hr/>
<img src="https://Digitalberg.net/blog/wp-content/themes/defaults/img/register.jpg" alt=""/>
<hr/>
<p><em class="after">This article includes insights and materials from various online sources. We acknowledge and value the efforts of all original authors, publishers, and websites. While every effort has been made to appropriately credit the source material, any unintentional oversight or omission does not imply a copyright infringement. All trademarks, logos, and images mentioned are the property of their respective owners. If you believe that any content used in this article infringes your copyright, please contact us immediately for review and swift action.</em></p>
<p><em class="after">This article aims solely for informational and educational purposes and does not violate the rights of copyright owners. If any copyrighted material has been unintentionally used without proper credit, we will promptly rectify this upon notification. Please note that the republication, redistribution, or reproduction of part or all of the contents in any form is prohibited without express written permission from the author and website owner. For permissions or further inquiries, please contact us.</em></p>

Related Posts