Skip to main content

Flask: APIs and Microservices

Flask: APIs and Microservices

Flask is a lightweight and flexible framework for building APIs and microservices, enabling modular, scalable, and maintainable web applications. Building on Flask Authentication, Flask User Registration, Flask User Login and Logout, Flask Password Hashing with Flask-Bcrypt, Flask Role-Based Access Control, Flask Protecting Routes with Flask-Login, Flask Managing User Sessions, Flask Using Flask Sessions, Flask Session Security, Flask Custom Session Backends, Flask Querying the Database, and Flask Relationships in Models, Flask integrates with tools like Flask-RESTful, Flask-JWT-Extended, and NumPy Array Operations for efficient API development. This tutorial explores Flask APIs and microservices, covering design, authentication, deployment, and practical use cases in modern web development.


01. Why Use Flask for APIs and Microservices?

Flask’s simplicity, extensibility, and minimal overhead make it ideal for building RESTful APIs and microservices. Key benefits include: - Modularity: Create independent services for specific functionalities. - Scalability: Deploy microservices individually to handle load. - Flexibility: Integrate with databases, authentication, and third-party services. - Lightweight: Minimal setup for rapid development. Flask APIs and microservices power applications like e-commerce platforms, content management systems, and IoT backends, leveraging Flask-Login and custom session backends for secure user management.

Example: Basic Flask API

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/users', methods=['GET'])
def get_users():
    users = [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
    return jsonify(users)

@app.route('/api/users', methods=['POST'])
def create_user():
    user = request.get_json()
    return jsonify({'id': 3, 'name': user['name']}), 201

if __name__ == '__main__':
    app.run(debug=True)

Output:

GET /api/users:
[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

POST /api/users with {"name": "Charlie"}:
{"id": 3, "name": "Charlie"}

Explanation:

  • jsonify - Converts Python dictionaries to JSON responses.
  • request.get_json - Parses JSON request bodies.
  • Basic RESTful endpoints for user data.

02. Key Techniques for Flask APIs and Microservices

Flask supports building robust APIs and microservices with extensions and best practices. Below is a summary of key techniques and their applications:

Technique Description Use Case
RESTful APIs with Flask-RESTful Structured API development Standardized endpoints
JWT Authentication Secure API access with tokens User authentication
Database Integration Persistent data storage Manage user data
Microservice Architecture Independent service deployment Scalable systems
API Documentation Auto-generate API specs Developer-friendly APIs


2.1 RESTful APIs with Flask-RESTful

Example: Flask-RESTful API

from flask import Flask
from flask_restful import Api, Resource, reqparse

app = Flask(__name__)
api = Api(app)

users = [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]

class UserResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('name', type=str, required=True, help='Name is required')

    def get(self, user_id):
        user = next((u for u in users if u['id'] == user_id), None)
        return user if user else ({'message': 'User not found'}, 404)

    def post(self):
        args = self.parser.parse_args()
        user = {'id': len(users) + 1, 'name': args['name']}
        users.append(user)
        return user, 201

class UserListResource(Resource):
    def get(self):
        return users

api.add_resource(UserListResource, '/api/users')
api.add_resource(UserResource, '/api/users/<int:user_id>')

if __name__ == '__main__':
    app.run(debug=True)

Output:

GET /api/users:
[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

POST /api/users with {"name": "Charlie"}:
{"id": 3, "name": "Charlie"}

GET /api/users/1:
{"id": 1, "name": "Alice"}

Explanation:

  • Flask-RESTful - Simplifies RESTful API development with resource classes.
  • reqparse - Validates and parses request data.

2.2 JWT Authentication with Flask-JWT-Extended

Example: JWT-Protected API

from flask import Flask, jsonify, request
from flask_jwt_extended import JWTManager, jwt_required, create_access_token
from flask_bcrypt import Bcrypt

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secure-key-123'
app.config['JWT_SECRET_KEY'] = 'jwt-secret-456'
jwt = JWTManager(app)
bcrypt = Bcrypt(app)

users = {'alice': bcrypt.generate_password_hash('password123').decode('utf-8')}

@app.route('/api/login', methods=['POST'])
def login():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    if username in users and bcrypt.check_password_hash(users[username], password):
        access_token = create_access_token(identity=username)
        return jsonify({'access_token': access_token})
    return jsonify({'message': 'Invalid credentials'}), 401

@app.route('/api/protected', methods=['GET'])
@jwt_required()
def protected():
    return jsonify({'message': 'Protected resource accessed'})

if __name__ == '__main__':
    app.run(debug=True)

Output:

POST /api/login with {"username": "alice", "password": "password123"}:
{"access_token": "eyJ..."}}

GET /api/protected with Authorization: Bearer <token>:
{"message": "Protected resource accessed"}

Explanation:

  • Flask-JWT-Extended - Secures APIs with JSON Web Tokens.
  • jwt_required - Protects routes requiring authentication.

2.3 Database Integration with SQLAlchemy

Example: API with SQLAlchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource, reqparse

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
api = Api(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)

class UserResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('name', type=str, required=True, help='Name is required')

    def get(self, user_id):
        user = User.query.get_or_404(user_id)
        return {'id': user.id, 'name': user.name}

    def post(self):
        args = self.parser.parse_args()
        user = User(name=args['name'])
        db.session.add(user)
        db.session.commit()
        return {'id': user.id, 'name': user.name}, 201

class UserListResource(Resource):
    def get(self):
        users = User.query.all()
        return [{'id': u.id, 'name': u.name} for u in users]

api.add_resource(UserListResource, '/api/users')
api.add_resource(UserResource, '/api/users/<int:user_id>')

with app.app_context():
    db.create_all()

if __name__ == '__main__':
    app.run(debug=True)

Output:

POST /api/users with {"name": "Charlie"}:
{"id": 1, "name": "Charlie"}

GET /api/users:
[{"id": 1, "name": "Charlie"}]

GET /api/users/1:
{"id": 1, "name": "Charlie"}

Explanation:

  • Integrates Flask-SQLAlchemy for persistent storage.
  • Uses Flask-RESTful for structured API endpoints.

2.4 Microservice Architecture

Example: User Microservice

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, jwt_required, create_access_token

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///micro_users.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'jwt-secret-789'
db = SQLAlchemy(app)
jwt = JWTManager(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(120), nullable=False)

@app.route('/api/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(username=data['username']).first()
    if user and user.password_hash == data['password']:  # Simplified check
        access_token = create_access_token(identity=user.id)
        return jsonify({'access_token': access_token})
    return jsonify({'message': 'Invalid credentials'}), 401

@app.route('/api/users/<int:user_id>', methods=['GET'])
@jwt_required()
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify({'id': user.id, 'username': user.username})

with app.app_context():
    db.create_all()
    db.session.add(User(username='alice', password_hash='hashed_password'))
    db.session.commit()

if __name__ == '__main__':
    app.run(debug=True, port=5001)

Output:

POST /api/login with {"username": "alice", "password": "hashed_password"}:
{"access_token": "eyJ..."}}

GET /api/users/1 with Authorization: Bearer <token>:
{"id": 1, "username": "alice"}

Explanation:

  • Implements a standalone user microservice with JWT authentication.
  • Runs on a separate port for modularity.

2.5 API Documentation with Flask-OpenAPI

Example: API Documentation with OpenAPI

from flask import Flask
from flasgger import Swagger

app = Flask(__name__)
swagger = Swagger(app)

@app.route('/api/users', methods=['GET'])
def get_users():
    """
    Get all users
    ---
    responses:
      200:
        description: A list of users
        schema:
          type: array
          items:
            type: object
            properties:
              id:
                type: integer
              name:
                type: string
    """
    users = [{'id': 1, 'name': 'Alice'}, {'id': 2, 'name': 'Bob'}]
    return jsonify(users)

if __name__ == '__main__':
    app.run(debug=True)

Output:

GET /api/users:
[{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

Access /apidocs for interactive Swagger UI

Explanation:

  • Flasgger - Generates interactive OpenAPI documentation.
  • Provides a Swagger UI at /apidocs for API exploration.

2.6 Insecure API Design

Example: Insecure API

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/users', methods=['GET'])
def get_users():
    # No authentication
    users = [{'id': 1, 'name': 'Alice', 'password': 'secret'}]
    return jsonify(users)

@app.route('/api/users', methods=['POST'])
def create_user():
    # No input validation
    user = request.get_json()
    return jsonify(user)

if __name__ == '__main__':
    app.run(debug=True)

Output:

GET /api/users:
[{"id": 1, "name": "Alice", "password": "secret"}]

POST /api/users with {"name": "Bob"}:
{"name": "Bob"}

Explanation:

  • Lacks authentication, exposing sensitive data.
  • No input validation, risking injection attacks.
  • Solution: Use JWT authentication and input validation.

03. Effective Usage

3.1 Recommended Practices

  • Use Flask-RESTful, Flask-JWT-Extended, and secure session backends.

Example: Comprehensive API with Microservice

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource, reqparse
from flask_jwt_extended import JWTManager, jwt_required, create_access_token
from flask_bcrypt import Bcrypt
from flasgger import Swagger

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///secure_api.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SECRET_KEY'] = 'secure-key-101'
app.config['JWT_SECRET_KEY'] = 'jwt-secret-202'
db = SQLAlchemy(app)
api = Api(app)
jwt = JWTManager(app)
bcrypt = Bcrypt(app)
swagger = Swagger(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(120), nullable=False)

class UserResource(Resource):
    """
    User Resource
    ---
    get:
      summary: Get user by ID
      responses:
        200:
          description: User details
    """
    @jwt_required()
    def get(self, user_id):
        user = User.query.get_or_404(user_id)
        return {'id': user.id, 'username': user.username}

class UserListResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('username', type=str, required=True, help='Username is required')
    parser.add_argument('password', type=str, required=True, help='Password is required')

    def post(self):
        args = self.parser.parse_args()
        if User.query.filter_by(username=args['username']).first():
            return {'message': 'Username exists'}, 400
        user = User(username=args['username'], password_hash=bcrypt.generate_password_hash(args['password']).decode('utf-8'))
        db.session.add(user)
        db.session.commit()
        return {'id': user.id, 'username': user.username}, 201

class LoginResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('username', type=str, required=True, help='Username is required')
    parser.add_argument('password', type=str, required=True, help='Password is required')

    def post(self):
        args = self.parser.parse_args()
        user = User.query.filter_by(username=args['username']).first()
        if user and bcrypt.check_password_hash(user.password_hash, args['password']):
            access_token = create_access_token(identity=user.id)
            return {'access_token': access_token}
        return {'message': 'Invalid credentials'}, 401

api.add_resource(UserListResource, '/api/users')
api.add_resource(UserResource, '/api/users/<int:user_id>')
api.add_resource(LoginResource, '/api/login')

with app.app_context():
    db.create_all()

if __name__ == '__main__':
    app.run(debug=True, port=5002)

Output:

POST /api/users with {"username": "frank", "password": "secure123"}:
{"id": 1, "username": "frank"}

POST /api/login with {"username": "frank", "password": "secure123"}:
{"access_token": "eyJ..."}}

GET /api/users/1 with Authorization: Bearer <token>:
{"id": 1, "username": "frank"}
  • Combines Flask-RESTful, Flask-JWT-Extended, and Flask-SQLAlchemy.
  • Includes OpenAPI documentation with Flasgger.
  • Secures endpoints with JWT and input validation.

3.2 Practices to Avoid

  • Avoid unauthenticated endpoints or unvalidated inputs.

Example: Insecure Microservice

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/api/data', methods=['POST'])
def post_data():
    # No validation or authentication
    data = request.get_json()
    return jsonify(data)

if __name__ == '__main__':
    app.run(debug=True)

Output:

POST /api/data with {"key": "value"}:
{"key": "value"}
  • No authentication or validation, risking security breaches.
  • Solution: Use JWT and validate inputs with reqparse.

04. Common Use Cases in Web Development

4.1 E-commerce Product API

Build a product microservice for an e-commerce platform.

Example: Product Microservice

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource, reqparse
from flask_jwt_extended import JWTManager, jwt_required

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///products.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'jwt-secret-303'
db = SQLAlchemy(app)
api = Api(app)
jwt = JWTManager(app)

class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)

class ProductResource(Resource):
    @jwt_required()
    def get(self, product_id):
        product = Product.query.get_or_404(product_id)
        return {'id': product.id, 'name': product.name, 'price': product.price}

class ProductListResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('name', type=str, required=True, help='Name is required')
    parser.add_argument('price', type=float, required=True, help='Price is required')

    @jwt_required()
    def get(self):
        products = Product.query.all()
        return [{'id': p.id, 'name': p.name, 'price': p.price} for p in products]

    @jwt_required()
    def post(self):
        args = self.parser.parse_args()
        product = Product(name=args['name'], price=args['price'])
        db.session.add(product)
        db.session.commit()
        return {'id': product.id, 'name': product.name, 'price': product.price}, 201

api.add_resource(ProductListResource, '/api/products')
api.add_resource(ProductResource, '/api/products/<int:product_id>')

with app.app_context():
    db.create_all()

if __name__ == '__main__':
    app.run(debug=True, port=5003)

Output:

POST /api/products with {"name": "Laptop", "price": 999.99} and Bearer token:
{"id": 1, "name": "Laptop", "price": 999.99}

GET /api/products with Bearer token:
[{"id": 1, "name": "Laptop", "price": 999.99}]

Explanation:

  • Secures product endpoints with JWT.
  • Uses SQLAlchemy for persistent storage.

4.2 User Authentication Microservice

Create a dedicated microservice for user authentication.

Example: Authentication Microservice

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager, create_access_token
from flask_restful import Api, Resource, reqparse
from flask_bcrypt import Bcrypt

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///auth.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['JWT_SECRET_KEY'] = 'jwt-secret-404'
db = SQLAlchemy(app)
api = Api(app)
jwt = JWTManager(app)
bcrypt = Bcrypt(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(120), nullable=False)

class LoginResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('username', type=str, required=True, help='Username is required')
    parser.add_argument('password', type=str, required=True, help='Password is required')

    def post(self):
        args = self.parser.parse_args()
        user = User.query.filter_by(username=args['username']).first()
        if user and bcrypt.check_password_hash(user.password_hash, args['password']):
            access_token = create_access_token(identity=user.id)
            return {'access_token': access_token}
        return {'message': 'Invalid credentials'}, 401

class RegisterResource(Resource):
    parser = reqparse.RequestParser()
    parser.add_argument('username', type=str, required=True, help='Username is required')
    parser.add_argument('password', type=str, required=True, help='Password is required')

    def post(self):
        args = self.parser.parse_args()
        if User.query.filter_by(username=args['username']).first():
            return {'message': 'Username exists'}, 400
        user = User(username=args['username'], password_hash=bcrypt.generate_password_hash(args['password']).decode('utf-8'))
        db.session.add(user)
        db.session.commit()
        return {'message': 'User created'}, 201

api.add_resource(LoginResource, '/api/login')
api.add_resource(RegisterResource, '/api/register')

with app.app_context():
    db.create_all()

if __name__ == '__main__':
    app.run(debug=True, port=5004)

Output:

POST /api/register with {"username": "grace", "password": "secure123"}:
{"message": "User created"}

POST /api/login with {"username": "grace", "password": "secure123"}:
{"access_token": "eyJ..."}

Explanation:

  • Provides authentication and registration endpoints.
  • Integrates with Flask-Bcrypt for secure password hashing.

Conclusion

Flask APIs and microservices, integrated with Flask-RESTful, Flask-JWT-Extended, Flask-SQLAlchemy, Flask Custom Session Backends, and NumPy Array Operations, enable scalable and secure web applications. Key takeaways:

  • Use Flask-RESTful for structured APIs and Flask-JWT-Extended for secure authentication.
  • Implement microservices for modularity and scalability.
  • Integrate with databases and document APIs with Flasgger.
  • Apply in e-commerce, authentication, or IoT for robust systems.

With these techniques, you can build Flask-based APIs and microservices that are secure, scalable, and developer-friendly, meeting the demands of modern web development!

Comments