
Create a REST API Using Flask on Ubuntu
Flask, Python’s minimalist web framework, offers developers a straightforward path to building robust REST APIs on Ubuntu servers. This guide walks you through creating production-ready Flask APIs, from initial setup to deployment considerations. You’ll learn essential patterns for API development, database integration, authentication strategies, and performance optimization techniques that scale with your application’s growth.
Understanding Flask for REST API Development
Flask operates on the WSGI principle, making it exceptionally lightweight compared to Django’s batteries-included approach. Its simplicity shines in API development where you need precise control over endpoints, request handling, and response formatting. Unlike FastAPI’s automatic OpenAPI generation or Django REST Framework’s serializer complexity, Flask gives you granular control over every aspect of your API.
The framework’s request-response cycle handles HTTP methods through decorators, making REST endpoint creation intuitive. Flask’s modular design allows you to add components like SQLAlchemy for database operations, Flask-JWT-Extended for authentication, or Flask-CORS for cross-origin requests only when needed.
Ubuntu Environment Setup and Flask Installation
Setting up Flask on Ubuntu requires Python 3.8+ and virtual environment management. Here’s the complete setup process:
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
Install Flask with essential extensions for API development:
pip install flask flask-sqlalchemy flask-migrate flask-cors flask-jwt-extended python-dotenv
For production deployments, add these packages:
pip install gunicorn psycopg2-binary redis
Create your project structure:
mkdir flask_rest_api && cd flask_rest_api
touch app.py config.py requirements.txt .env .gitignore
Building Your First REST API
Start with a basic Flask application structure that follows REST conventions:
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': 'Username and email required'}), 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'])
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 run your API:
export FLASK_APP=app.py
flask db init
flask db migrate -m "Initial migration"
flask db upgrade
python app.py
Test your endpoints using curl:
curl -X POST http://localhost:5000/api/users \
-H "Content-Type: application/json" \
-d '{"username": "testuser", "email": "test@example.com"}'
curl http://localhost:5000/api/users
Advanced API Features and Authentication
Production APIs require authentication, input validation, and error handling. Here’s an enhanced version with JWT authentication:
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())
Database Integration and Migration Strategies
Flask-SQLAlchemy provides excellent PostgreSQL integration for production environments. Configure PostgreSQL connection:
sudo apt install postgresql postgresql-contrib
sudo -u postgres createuser --interactive
sudo -u postgres createdb flask_api_db
Update 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
Handle database migrations properly:
flask db migrate -m "Add password field to users"
flask db upgrade
For complex 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]
}
Performance Optimization and Caching
Implement Redis caching for frequently accessed data:
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]
Add database query optimization:
# 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
})
Framework Comparison and Use Cases
Framework | Setup Complexity | Performance | Learning Curve | Best For |
---|---|---|---|---|
Flask | Low | Good | Gentle | Small to medium APIs, microservices |
FastAPI | Low | Excellent | Moderate | High-performance APIs, async operations |
Django REST | High | Good | Steep | Complex applications, admin interfaces |
Express.js | Low | Excellent | Moderate | JavaScript ecosystem, real-time apps |
Flask excels in scenarios requiring:
- Rapid prototyping and development
- Custom authentication systems
- Integration with existing Python codebases
- Microservices architecture
- Educational projects and learning
Production Deployment with Gunicorn and Nginx
Configure Gunicorn for production serving:
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 as 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
Common Pitfalls and Troubleshooting
Flask development often encounters these issues:
**Database Connection Problems**: Always use connection pools and handle database errors gracefully:
from sqlalchemy.exc import OperationalError
@app.errorhandler(OperationalError)
def handle_db_error(error):
db.session.rollback()
return jsonify({'error': 'Database connection failed'}), 500
**CORS Issues**: Configure CORS properly for frontend integration:
from flask_cors import CORS
CORS(app, origins=['http://localhost:3000', 'https://yourdomain.com'])
**Memory Leaks**: Close database sessions and use connection pooling:
@app.teardown_appcontext
def close_db(error):
if hasattr(g, 'db_connection'):
g.db_connection.close()
**JWT Token Issues**: Implement proper token refresh mechanisms and handle expired tokens:
@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
return jsonify({'error': 'Token has expired'}), 401
Performance monitoring becomes crucial at scale. Implement 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)
Flask’s flexibility makes it ideal for API development when you need control over every component. The framework’s minimalist approach means you add complexity only when necessary, making it perfect for both learning REST concepts and building production-grade APIs. For comprehensive Flask documentation and advanced patterns, visit the official Flask documentation and explore the Flask-SQLAlchemy guide for database integration best practices.

This article incorporates information and material from various online sources. We acknowledge and appreciate the work 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 constitute 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 upon your copyright, please contact us immediately for review and prompt action.
This article is intended for informational and educational purposes only and does not infringe on the rights of the copyright owners. If any copyrighted material has been used without proper credit or in violation of copyright laws, it is unintentional and we will rectify it promptly upon notification. Please note that the republishing, 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.