REST APIs with Flask

Flask is a lightweight Python web framework that allows developers to build REST APIs with ease. By using Flask along with Flask extensions like Flask-RESTful and Flask-SQLAlchemy, we can create efficient and scalable RESTful APIs.

---

1. Installing Flask and Required Extensions

Before we start, install Flask and the required extensions using pip:

pip install flask flask-restful flask-sqlalchemy flask-migrate

---

2. Setting Up a Basic Flask API

Let’s create a simple Flask API that returns a JSON response.

app.py

from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/api/hello', methods=['GET'])
def hello():
    return jsonify({"message": "Hello, World!"})

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

• We define a route /api/hello that responds with a JSON message.
• The jsonify function ensures the response is returned in JSON format.

Run the app:

python app.py

Test the API in a browser or using Postman at:

http://127.0.0.1:5000/api/hello

---

3. Creating a RESTful API with Flask-RESTful

Flask-RESTful makes building APIs easier by providing a class-based approach.

Setting Up Flask-RESTful

Modify app.py to use Flask-RESTful:

from flask import Flask
from flask_restful import Resource, Api

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

class HelloWorld(Resource):
    def get(self):
        return {"message": "Hello, Flask-RESTful!"}

api.add_resource(HelloWorld, '/api/hello')

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

• We define a class HelloWorld that inherits from Resource.
• The get() method handles GET requests.

Test the API at:

http://127.0.0.1:5000/api/hello

---

4. Implementing CRUD Operations

We’ll now create a REST API for managing users with CRUD (Create, Read, Update, Delete) operations.

4.1 Setting Up Database with Flask-SQLAlchemy

Modify app.py to include database configuration:

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_restful import Api, Resource

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(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)

db.create_all()  # Create database tables

• We define a User model with id, name, and age columns.
• db.create_all() creates the table in the SQLite database.

---

4.2 Defining CRUD API Endpoints

Now, let’s create an API to Add, Retrieve, Update, and Delete users.

Create and Retrieve Users

class UserResource(Resource):
    def get(self):
        users = User.query.all()
        return jsonify([{"id": user.id, "name": user.name, "age": user.age} for user in users])

    def post(self):
        data = request.get_json()
        new_user = User(name=data['name'], age=data['age'])
        db.session.add(new_user)
        db.session.commit()
        return jsonify({"message": "User added successfully!"})

• GET method retrieves all users.
• POST method adds a new user.

Update and Delete Users

class UserByID(Resource):
    def get(self, user_id):
        user = User.query.get(user_id)
        if user:
            return jsonify({"id": user.id, "name": user.name, "age": user.age})
        return jsonify({"message": "User not found"}), 404

    def put(self, user_id):
        user = User.query.get(user_id)
        if user:
            data = request.get_json()
            user.name = data.get('name', user.name)
            user.age = data.get('age', user.age)
            db.session.commit()
            return jsonify({"message": "User updated successfully"})
        return jsonify({"message": "User not found"}), 404

    def delete(self, user_id):
        user = User.query.get(user_id)
        if user:
            db.session.delete(user)
            db.session.commit()
            return jsonify({"message": "User deleted successfully"})
        return jsonify({"message": "User not found"}), 404

• GET /api/user/<id> – Retrieve user by ID.
• PUT /api/user/<id> – Update user.
• DELETE /api/user/<id> – Delete user.

Register the Resources

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

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

---

5. Testing the API

5.1 Create a User (POST)

Endpoint:

POST http://127.0.0.1:5000/api/users

Body (JSON):

{
    "name": "John Doe",
    "age": 30
}

5.2 Retrieve All Users (GET)

Endpoint:

GET http://127.0.0.1:5000/api/users

Response:

[
    {
        "id": 1,
        "name": "John Doe",
        "age": 30
    }
]

5.3 Retrieve a User by ID (GET)

Endpoint:

GET http://127.0.0.1:5000/api/user/1

Response:

{
    "id": 1,
    "name": "John Doe",
    "age": 30
}

5.4 Update a User (PUT)

Endpoint:

PUT http://127.0.0.1:5000/api/user/1

Body (JSON):

{
    "name": "John Smith",
    "age": 35
}

5.5 Delete a User (DELETE)

Endpoint:

DELETE http://127.0.0.1:5000/api/user/1

---

6. Adding Authentication with Flask-JWT

For authentication, we use Flask-JWT-Extended.

pip install flask-jwt-extended

Modify app.py:

from flask_jwt_extended import JWTManager, create_access_token, jwt_required

app.config['JWT_SECRET_KEY'] = 'your_secret_key'
jwt = JWTManager(app)

class Login(Resource):
    def post(self):
        data = request.get_json()
        if data['username'] == 'admin' and data['password'] == 'password':
            token = create_access_token(identity=data['username'])
            return jsonify({"access_token": token})
        return jsonify({"message": "Invalid credentials"}), 401

api.add_resource(Login, '/api/login')

class ProtectedResource(Resource):
    @jwt_required()
    def get(self):
        return jsonify({"message": "Access granted!"})

api.add_resource(ProtectedResource, '/api/protected')

• Login with /api/login to receive a JWT token.
• Use this token to access /api/protected.