Deploy a Python flask server using Google Cloud Run

From a basic skeleton to a ready-to-deploy server.

just about the time to finish your tea

This post shows how to deploy a minimal Python Flask application to production from zero (really). On Cloud Run.

It could be the starting point of any server project.

Minimum installation. Minimum code.

At first, we will create a minimal Flask server, just able to respond to an HTTP call.

If you are interested by Nodejs, I also made an article about doing the same thing in Nodejs using Cloud Compute Engine.

That’s a warm up structure.

We will then move to the most important part, create a useable skeleton ready for production, including:

• Ready-to-deploy structure
• Generic, relevant responses
• Generic api calls method
• Generic and efficient logging
• Advice on input validation

Docker & Cloud Run: Prerequisites

• Have Docker installed
• Docker daemon running.
• gcloud CLI installed

Minimal structure — Basic skeleton

Run the following commands:

> mkdir flask_skeleton
> cd flask_skeleton
> mkdir utils
> touch utils/utils.py
> mkdir blueprints
> touch blueprints/activities.py
> touch requirements.txt
> touch main.py
> touch Dockerfile
> touch config.py
> touch .dockerignore

It will create the files and folders that we need and will need to have an efficient structure.

We now have this structure:

Inside requirements.txt, add the following line:

Flask==2.2.3 
gunicorn==20.1.0

Inside main.py, add the following line:

import os
from flask import Flask, jsonify, Response

def create_app():
  app = Flask(__name__)
  # Error 404 handler
  @app.errorhandler(404)
  def resource_not_found(e):
    return jsonify(error=str(e)), 404
  # Error 405 handler
  @app.errorhandler(405)
  def resource_not_found(e):
    return jsonify(error=str(e)), 405
  # Error 401 handler
  @app.errorhandler(401)
  def custom_401(error):
    return Response("API Key required.", 401)
  
  @app.route("/ping")
  def hello_world():
     return "pong"
  
  return app
  
app = create_app()

if __name__ == "__main__":
  #    app = create_app()
  print(" Starting app...")
  app.run(host="0.0.0.0", port=5000)

Inside Dockerfile, add:

FROM python:3.10-slim

ENV PYTHONUNBUFFERED True

ENV APP_HOME /app

ENV PORT 5000

WORKDIR $APP_HOME

COPY . ./

RUN pip install --no-cache-dir -r requirements.txt

CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 main:app

This Dockerfile contains all the commands that we could manually call on the command line to assemble the image. Having it inside a Dockerfile automate the image creation.

Inside .dockerignore, add:

Dockerfile
README.md
*.pyc
*.pyo
*.pyd
__pycache__
.pytest_cache

Easy peazy.

And…

You are set.

If you are new to Docker, it might seems weird. But with this, everything’s ready to deploy a minimal server to Cloud Run.

Run the following command:

# This show where to find your project
PATH_FOLDER=$(pwd)

# This will build your container
docker build --tag flask_skeleton .

# Run the container
docker run -p 5000:5000 flask_skeleton

If your Docker is configured correctly, you should have 4 lines displayed:

You can then simply go to a browser and run http://localhost:5000/ping and get your “pong”.

Nice!

If you don’t have this result. You could have issues with your Docker config/environment variables.

Do go to the next sections before this is fixed. Post a comment if you struggle.

Ready-to-deploy skeleton

With the previous structure, we are not efficient when developing.

If that was for a test, we are ready to deploy but if we want to go further, there are some things to fix.

Plus, we still have a long way before doing an API call.

Plus, once deployed, the logs won’t be easily readable inside Cloud Run console.

Plus we haven’t seen how to deploy this container to Cloud Run ;)

Skeleton V2

We will add some methods and some structure that will definitely be useful for production:

• Ready-to-deploy structure
• Generic, relevant responses
• Generic api calls method
• Generic Execution ID for Cloud Run logs
• Advice on input validation

We will create 2 routes:

• GET /version: Returning the current version of the server
• POST /activities: A POST request sending back our POST request (that ain’t much… but it’s enough to test POST request and input validation)

After implementing the structure and the functions, it will be super easy to add another route/another API call…

Version and time

A very good practice when developing a server (especially at the beginning), is to always return the server version and the time the response was sent.

Doing so, we are assured that we won’t spend an hour with a stupid mistake from an older version.

It’s also very useful with Cloud Run. Because with Cloud Run, we are able to deploy a version that will be ran by 10%, 50% of the traffic.

Knowing what version was executed could help, don’t you think?

Sooooo

First, we need to create the version variable inside the config.py file:

VERSION = "0.1"

Inside main.py , add this route:

@app.route("/version", methods=["GET"], strict_slashes=False)
def version():
  response_body = {
      "success": 1,
  }
  return jsonify(response_body)

This doesn’t return the version… True!

Because, and that’s the magic of Flask, we will be adding the version at the end of an execution. So every API call will receive the version!

Add these lines into main.py:

import config
import json
import time

...

@app.after_request
def after_request(response):
    if response and response.get_json():
        data = response.get_json()

        data["time_request"] = int(time.time())
        data["version"] = config.VERSION

        response.set_data(json.dumps(data))

    return response

We can rebuild our image and restart it:

> docker build - tag flask_skeleton .
> docker run -p 5000:5000 flask_skeleton

When calling http://localhost:5000/version, we receive some data that might be useful even with a large server: {“success”: 1, “time_request”: 1679500000, “version”: “0.1”}.

Generic API call function

To handle API calls, we will use a generic function.

This function is handling errors for us and can hide some complexity in case of specific needs from your own server.

In this example, having a generic function is not so relevant. But we might have a more complex structure for some API. Generalising the authentication, for instance, could be efficient.

This function will be in ./utils/utils.py.

import requests

def generic_api_requests(method, url, payload={}, params={}):
    print("CURRENT REQUEST : ", method, url, payload)

    try:
        response = requests.request(
            method,
            url,
            json=payload,
            params=params,
        )

        json_response = response.json()

        print( "RESPONSE SUCCESS")

        return 1, json_response

    except Exception as e:
        print("RESPONSE ERROR :", e)
        return 0, e

Don’t forget to add requests in requirements.txt

requests==2.28.2

Simple Blueprints

If you got here, it’s because you are looking for a skeleton able to become a large server!

But if this cute skeleton becomes a larger server, you will have multiple components, entities working together with API calls from everywhere going anywhere.

We need to split the components into Blueprints.

Flask will dispatch requests and generate URLs from one endpoint to another.

At the beginning of this tutorial, we created a file blueprints/activities.py.

With this file, we will receive a POST API call, handle it, call an external API (with our generic function) and respond to the user.

import config
import json

from flask import Blueprint, jsonify, request

from utils.utils import generic_api_requests

activities = Blueprint(name="activities", import_name=__name__)


@activities.route("/", methods=["POST"], strict_slashes=False)
def create_activity():
    try:

        request_body = request.get_json()

        is_success, response = generic_api_requests(
            "post", config.URL_ACTIVITIES, request_body
        )

        response_body = {
            "success": is_success,
            "data": response["json"] if is_success else {"message": str(response)},
        }

        return jsonify(response_body)

    except Exception as error:

        response_body = {
            "success": 0,
            "data": {"message": "Error : {}".format(error)},
        }

        return jsonify(response_body), 400

That Blueprint’s mission is only to receive an API call, call an external API and respond to the initial caller.

In config.py, set the URL_ACTIVITIES.

URL_ACTIVITIES = "https://httpbin.org/post"

And finally, at the beginning of main.py, declare the Blueprint.

from blueprints.activities import activities

def create_app():
    app = Flask(__name__)
    app.register_blueprint(activities, url_prefix="/api/v1/activities")

We can now rebuild what we have just done, and do a POST request to http://0.0.0.0:5000/api/v1/activities { “test”:”1" }

I hope you could get until here!

If not please leave a message so I can improve this step-by-step tutorial.

The last essential thing

This tutorial is being quite long and I am not sure about it’s popularity 😅

So I won’t go into details for input validation, but if you clone the git repository of this project, the POST route will have a skeleton of input validation made with pydantic.

But there is an important part I kept for the end… Logging!

When we deploy a Cloud Run instance, there is no execution ID, natively.

Our logs will be mixed up and to be honest, it will be quite messy.

We have to create one, so when a request is done on the server, we are able to find the whole execution related to this request.

This is very useful in case of error.

I will show you, after deployment, how to use it with Cloud Logging.

Let’s see first how to implement it:

Inside main.py, add these 2 lines and the following function:

from flask import Flask, jsonify, Response, g, request
import uuid
...
...
...
  @app.before_request
  def before_request_func():
    execution_id = uuid.uuid4()
    g.start_time = time.time()
    g.execution_id = execution_id

    print(g.execution_id, "ROUTE CALLED ", request.url)

And you just created an execution id ;)

Now, every execution has a global (in a flask point of view) variable named execution_id. We can use it in every file of this server.

Do you see my “print“ in the piece of code?

Right after the print(, there is an g.execution_id. This global variable contains the execution id accessible from any file of this server.

What we have to do to get an execution id at every print, is to add this g.execution_id inside every print!

It’s as simple as that.

Also, we need to import g in every file where it is used : from flask import g

After this small change, our logs look like that:

Trust me… you saved hours debugging.

And because you saved hours, you have to clap me multiple times !

Lift off!

Our server is running locally. It’s time to deploy it on Cloud Run.

If you have gcloud CLI installed, it’s super easy.

If not, install gcloud CLI, it will be super easy.

Once in the flask_skeleton folder, run:

gcloud run deploy flask-skeleton --region=europe-west2 --source=$(pwd) --allow-unauthenticated

Note that this service is publicly accessible. Don’t forget to delete it once you are done testing.

After deployment, we get that:

Click flask-skeleton.

At the very top, you have the service URL:

You can magically replace the localhost we had before by that URL and get the server version or create an activity!

YEAH!

It’s that easy!

How about monitoring and logs?

We created an execution ID, let’s see how it goes!

Go on Cloud Run dashboard in the “Logs” tab.

execution id youhou

And so what? We have execution id and?

Imagine you have 100 logs per execution.

And imagine after a while, the execution crash.

You have no way, natively, to get the entire execution to understand the source of the crash.

Your best chance is to find, roughly, when did the crash happened and scroll through all logs.

If there was multiple executions at the same time… better give up.

With execution id, printed with every log, copy it, go to Cloud Logging and past this code:

resource.type = "cloud_run_revision"
resource.labels.service_name = "flask-skeleton"
resource.labels.location = "europe-west2"
 severity>=DEFAULT
 "e85b703d-1d85-4628-add9-20dc582f7177"

You now have all the prints linked to an execution!

Brilliant!

Code

The full code, and input validation is accessible here:

GitHub - BerangerN/flask-skeleton: Flask Skeleton, ready to be deployed to Cloud Run

Thanks?

Thanks.