fbpx

Writing a Python microservice

In this post we will write a minimal Python microservice, this time using bcrypt, flask, pyjwt and sqlalchemy.

Our requirements will be

bcrypt
flask
Flask-SQLAlchemy
pyjwt
  • bcrypt: package to hash our passwords and verify them
  • Flask: for our http application layer
  • Flask-SQLAlchemy: the integration of SQLAlchemy with flask, SQLAlquemy will be used as a SQL toolkit and ORM (Object Relational Mapper) to easily use our database.
  • pyjwt: Json web tokens package will be used to authenticate/authorize calls to the microservice

We can install this requirements with the command

pip install -r requirements.txt


Now, we have to write a microservice with a secured endpoint, our endpoint will return a json list with all the users in our system. Our user model will be

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String, unique=True, nullable=False)
    email = db.Column(db.String, unique=True, nullable=False)
    password = db.Column(db.String, nullable=False)
    role = db.Column(db.String, nullable=False)

    @classmethod
    def create_user(self, username, email, password, role='user'):
        """Sample method to create user with hashed password."""
        encrypted_password = bcrypt.hashpw(password.encode(), app.secret_key)
        user = User(username=username,
                    email=email,
                    password=encrypted_password,
                    role=role
                    )
        db.session.add(user)
        db.session.commit()
        return user

    def get_jwt(self):
        return jwt.encode({
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'role': self.role
        }, app.secret_key)

    def as_dict(self):
        return {
            c.name: getattr(self, c.name)
            for c in self.__table__.columns
            if c.name != 'password'
        }


We have some important stuffs here, the create_user method that creates the user saving the password hashed which is a very important sutff. We have the get_jwt method to easily obtain a json token from an user instance, and we have the as_dict method which transform the user model to dictionary excluding the password field.

The endpoint that we will need should look like this

@app.route('/users', methods=['GET'])
@check_jwt
@user_is_admin
def users():
    users = User.query.all()
    return jsonify([user.as_dict() for user in users])


Seems very easy isn’t it? but this endpoint have a lot of logic in it’s decorators. The decorator @check_jwt verifies the token provided by the request and authenticates the user, the @user_is_admin decorator will check the authenticated user has admin role.

The decorators looks like

def check_jwt(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        auth = request.headers.get('Authorization')

        if not auth:
            return 'Unauthorized', 401

        try:
            _, token = auth.split(' ')
            data = jwt.decode(token, app.secret_key)
            g.user_token = data
            g.user = User.query.filter_by(id=data['id']).one()

        except jwt.exceptions.InvalidSignatureError:
            app.logger.info('jwt Invalid Signature')
            return 'Unauthorized', 401

        except (ValueError, jwt.exceptions.DecodeError):
            app.logger.info('bad token')
            return 'Bad Request', 400

        return f(*args, **kwargs)
    return decorated_function


def user_is_admin(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user' in g and g.user.role == 'admin':
            return f(*args, **kwargs)
        return 'Unauthorized', 401
    return decorated_function


This decorators securizes our endpoints, in the g proxy we will have the user attribute with our authenticated User for the request because @check_jwt decorator is being used.

The Python microservice

The complete mycroservice is in the next code snippet

from functools import wraps

import bcrypt
from flask import Flask, jsonify, g, request
from flask_sqlalchemy import SQLAlchemy
import jwt


app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///db.sqlite"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
# generated with bcrypt.gensalt()
app.secret_key = b'$2b$12$B8wYmzgMNfoRQXZg8EJ1GO'

db = SQLAlchemy(app)
# compatibility with standard wsgi application name
application = app


class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String, unique=True, nullable=False)
    email = db.Column(db.String, unique=True, nullable=False)
    password = db.Column(db.String, nullable=False)
    role = db.Column(db.String, nullable=False)

    @classmethod
    def create_user(self, username, email, password, role='user'):
        """Sample method to create user with hashed password."""
        encrypted_password = bcrypt.hashpw(password.encode(), app.secret_key)
        user = User(username=username,
                    email=email,
                    password=encrypted_password,
                    role=role
                    )
        db.session.add(user)
        db.session.commit()
        return user

    def get_jwt(self):
        return jwt.encode({
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'role': self.role
        }, app.secret_key)

    def as_dict(self):
        return {
            c.name: getattr(self, c.name)
            for c in self.__table__.columns
            if c.name != 'password'
        }


def check_jwt(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        auth = request.headers.get('Authorization')

        if not auth:
            return 'Unauthorized', 401

        try:
            _, token = auth.split(' ')
            data = jwt.decode(token, app.secret_key)
            g.user_token = data
            g.user = User.query.filter_by(id=data['id']).one()

        except jwt.exceptions.InvalidSignatureError:
            app.logger.info('jwt Invalid Signature')
            return 'Unauthorized', 401

        except (ValueError, jwt.exceptions.DecodeError):
            app.logger.info('bad token')
            return 'Bad Request', 400

        return f(*args, **kwargs)
    return decorated_function


def user_is_admin(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if 'user' in g and g.user.role == 'admin':
            return f(*args, **kwargs)
        return 'Unauthorized', 401
    return decorated_function


@app.route('/users', methods=['GET'])
@check_jwt
@user_is_admin
def users():
    users = User.query.all()
    return jsonify([user.as_dict() for user in users])


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


In this example I have used the SQLite to easily setup a database system, SQLAlchemy also works with Mysql or Postgres. The service can be executed with python service.py but before that, we need to insert an User and generate a jwt for testing our endpoint. This can be done with the next script


from service import db
from service import User

# To clear database
# db.drop_all()
db.create_all()

# create sample user
user = User.create_user('someuser', 'mail@example.com', '11111111', 'admin')
print(user.get_jwt())


This script should be placed in the same folder as the flask service because, it imports variables from the service script. Running this script we will obtain something like

python sample.py
b'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MSwidXNlcm5hbWUiOiJzb21ldXNlciIsImVtYWlsIjoibWFpbEBleGFtcGxlLmNvbSIsInJvbGUiOiJhZG1pbiJ9.YvDeu9MbKlyqm4oOmM3cUOgNGOwulXuPACrWP-5Y-Hw'


For this example, we need to take the token eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MSwidXNlcm5hbWUiOiJzb21ldXNlciIsImVtYWlsIjoibWFpbEBleGFtcGxlLmNvbSIsInJvbGUiOiJhZG1pbiJ9.YvDeu9MbKlyqm4oOmM3cUOgNGOwulXuPACrWP-5Y-Hw for our test. Now let’s run the service and do a sample call.

We can run the service with

python service.py
 * Serving Flask app "service" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Running on http://0.0.0.0:8080/ (Press CTRL+C to quit)
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 198-167-948

And we can send a sample call with

samples/blog/microservice_flask » export TOKEN="eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpZCI6MSwidXNlcm5hbWUiOiJzb21ldXNlciIsImVtYWlsIjoibWFpbEBleGFtcGxlLmNvbSIsInJvbGUiOiJhZG1pbiJ9.YvDeu9MbKlyqm4oOmM3cUOgNGOwulXuPACrWP-5Y-Hw"
samples/blog/microservice_flask ‹master*› » curl -H "Authorization: Bearer $TOKEN" http://localhost:8080/users
[
  {
    "email": "mail@example.com",
    "id": 1,
    "role": "admin",
    "username": "someuser"
  }
]
samples/blog/microservice_flask »

Conclusion

We now have a Python microservice with

  • Secured endpoint
  • Decorators for authentication and authorization of users
  • Saving users with password hashing

In spite of all, this example needs more work, you could try to extend it by adding

  • Another endpoint with not user_admin restriction
  • Jwt token expiration
  • Login endpoint: Receives user credentials, check’s the user and returns a new jwt

Leave a Reply

Your email address will not be published. Required fields are marked *

%d bloggers like this: