Connecting to a Database in Flask

Connecting to a MySQL/MariaDB Database

To connect to a MySQL database, you first need to install the mysql-connector-python library. Use the following command to install this library:

pip install mysql-connector-python

Then, to connect to the MySQL database, use the following code:

from flask import Flask, jsonify, render_template
import mysql.connector
import logging
import os

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

DB_USER = os.environ.get('DB_USER', '')
DB_PASSWORD = os.environ.get('DB_PASSWORD', '')
DB_HOST = os.environ.get('DB_HOST', '')
DB_PORT = os.environ.get('DB_PORT', '')
DB_NAME = os.environ.get('DB_NAME', '')

def test_mysql_connection():
    try:
        conn = mysql.connector.connect(
            host=DB_HOST,
            user=DB_USER,
            password=DB_PASSWORD,
            database=DB_NAME,
            port=DB_PORT
        )
        logging.info("Successfully connected to MySQL")
        conn.close()
        return True
    except Exception as e:
        logging.error(f"MySQL connection failed: {e}")
        return False

@app.route('/')
def home():
    if test_mysql_connection():
        return jsonify({"message": "Successfully connected to MySQL"})
    return jsonify({"message": "Failed to connect to MySQL"})

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

In this code, we first import the necessary libraries. Then, we define the test_mysql_connection function, which checks the connection to the MySQL database. This function uses environment variables for the database settings. It checks the connection to the database and returns an appropriate message to the user upon success.

In the home function, we use the test_mysql_connection function to check the connection to the MySQL database. If successful, an appropriate message is returned.

To run this program in a container, you need to add the MySQL client app to run.sh:

apt-get install default-mysql-client

Connecting to a Redis Database

To connect to a Redis database, you first need to install the redis library. Use the following command to install this library:

pip install redis

Then, to connect to the Redis database, use the following code:

import os
from flask import Flask, request, jsonify
import redis
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

REDIS_PASS = os.environ.get('REDIS_PASS')
REDIS_HOST = os.environ.get('REDIS_HOST')
REDIS_PORT = os.environ.get('REDIS_PORT')

def check_redis_connection():
    try:
        redis_client = redis.Redis(host=REDIS_HOST, port=int(REDIS_PORT), password=REDIS_PASS)
        redis_client.ping()
        logging.info("Successfully connected to Redis")
        return True
    except redis.ConnectionError as e:
        logging.error(f"Failed to connect to Redis: {str(e)}")
        return False

@app.route('/', methods=['GET'])
def home():
    if check_redis_connection():
        return jsonify({"message": "Successfully connected to Redis"})
    return jsonify({"message": "Failed to connect to Redis"})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

tip

You can also use the relevant libraries to connect to other databases. Use the following commands to install the libraries for other databases:

# PostgreSQL
pip install psycopg2

# MongoDB
pip install pymongo

# Elasticsearch
pip install elasticsearch