Inter-Pod Communication and External Access

Our Flask application will need to communicate with the PostgreSQL database. To do this, we need to create a Service that exposes the PostgreSQL Pod to the cluster. We will use a ClusterIP Service to expose the database internally within the cluster.

cat < kubernetes/postgres-service.yaml
apiVersion: v1
kind: Service
metadata:
  # Sets service name
  name: postgres
  namespace: postgres
  labels:
    # Labels and Selectors
    app: postgres
spec:
  # Sets service type
  type: ClusterIP
  selector:
    app: postgres
  ports:
    - name: postgres
      protocol: TCP
      port: 5432
      targetPort: 5432
EOF

Create the service:

kubectl apply -f kubernetes/postgres-service.yaml

We are also going to modify the application code so that it connects to the PostgreSQL database and stores every to-do item externally. We also need to install some dependencies in our virtual environment.

Make sure to activate the virtual environment:

workon stateful-flask
# or mkvirtualenv stateful-flask

Then update the required dependencies:

cat < app/requirements.txt
Flask==3.0.1
Flask-SQLAlchemy
psycopg2-binary
Flask-Migrate
EOF

Now, update the application code to use PostgreSQL:

cat <<EOF > app/app.py
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

# Instantiate the Flask application
app = Flask(__name__)

# Configure the database connection
app.config['SQLALCHEMY_DATABASE_URI'] = (
    'postgresql://' # Database type
    'stateful-flask-user:stateful-flask-password@' # User and password
    'postgres.postgres.svc.cluster.local:5432/' # Host and port
    'stateful-flask-db' # Database name
)

# Configure SQLAlchemy
db = SQLAlchemy(app)
migrate = Migrate(app, db)

# Define the Task model
class Task(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(80), nullable=False)
    description = db.Column(db.String(200))

# The GET endpoint to retrieve all tasks
@app.route('/tasks', methods=['GET'])
def get_tasks():
    tasks = Task.query.all()
    return jsonify(
      {
        'tasks': [
          {
            'id': task.id,
            'title': task.title,
            'description': task.description
          }
          for task in tasks
        ]
      }
    )

# The POST endpoint to create a new task
# and store it in the database
@app.route('/tasks', methods=['POST'])
def create_task():
    data = request.get_json()
    title = data['title']
    description = data['description']
    task = Task(title=title, description=description)
    db.session.add(task)
    db.session.commit()
    return jsonify(
      {
        'task': {
          'id': task.id, 'title': task.title, 'description': task.description
        }
      }
    )

if __name__ == '__main__':
    app.run(
      debug=True,
      host='0.0.0.0',
      port=

Cloud-Native Microservices With Kubernetes - 2nd Edition

A Comprehensive Guide to Building, Scaling, Deploying, Observing, and Managing Highly-Available Microservices in Kubernetes

Enroll now to unlock all content and receive all future updates for free.

Unlock now ($31.99) Learn More

Previous Next