How to connect to a Postgres database with Python

Ello ello ello my fellow engineers!

This post is a supplement to Episode 3 and Episode 4 of the How to build a Scalable Python Application series on YouTube (you can see Episode 2 here: https://youtu.be/FMUsVwcBOVY)

We’ll be going over how to setup your Python application with a Postgres database connection (created in Episode 2) using Flask and SQLAlchemy.

Firstly, for this project I’m using Pipenv, you can use whatever virtual environment and dependency management you like, but personally I like to use Pipenv.

The documentation for Pipenv is here: https://pipenv.pypa.io/en/latest/.

I wrote a blog about it a long time ago here: https://medium.com/analytics-vidhya/a-tool-all-python-developers-should-be-using-5d547bfb45b7

The easiest way to install Pipenv on Mac is using Homebrew:

brew install pipenv

Next for our project we are going to have the following Pipfile:

[[source]]
name = "pypi"
url = "https://pypi.org/simple"
verify_ssl = true

[dev-packages]
pytest="*"
pytest_mock="*"

[packages]
clothes-shop = {editable = true, path = "."}
flask="*"
requests="*"
sqlalchemy="*"
flask-sqlalchemy="*"
pg8000="==1.16.5"

[requires]
python_version = "3.9"

We’ll be using flask for the application and SQLAlchemy for the database ORM (Objected Related Mapping) and pg8000 as the PostgreSQL engine. Requests will be used in a later tutorial (so ignore for now).

Documentation:

Flask: https://flask.palletsprojects.com/en/1.1.x/

SQLAlchemy: https://docs.sqlalchemy.org/en/13/

pg8000: https://pypi.org/project/pg8000/

We’ll be using pg8000, which is a pure Python interface for PostgreSQL database engine.

Fun fact: it’s called pg8000 because it is believed to be the 8000th interface with PostgreSQL for Python.

Unfortunately pg8000 is not full supported until SLQAlchemy 1.4 so we have to pin the version to 1.16.5 (see more on this thread https://github.com/sqlalchemy/sqlalchemy/issues/5645)

This pipfile will need to be on the same level as your project, in our case we are creating a clothes_shop directory to store our application_setup.py

Our project structure will look like this:

ClothesShop/
├── clothes_shop/
│   ├── test/
│   └── application_setup.py
│
├── README.md
├── setup.py
├── Pipfile
└── Pipfile.lock

We will also need to create a setup.py before we install our dependencies you can create aREADME.md and leave it empty for now:

"""setup.py"""
from setuptools import setup

with open('README.md', encoding="utf-8") as readme_file:
    readme = readme_file.read()

setup(
    name='clothes_shop',
    description='Clothes shop project for YouTube',
    author='Alexander V',
    author_email='alex.v.engineering@gmail.com',
    long_description=readme,
    test_suite='clothes_shop/test/'
)

If you want to install these dependencies using pipenv:

pipenv install -d

-d for development packages included. You will notice that a pipfile.lock file has been created as well.

If you followed the previous episode and have Postgres set up. You can add a password by doing the below:

Firstly, start psql by typing this into your terminal:

psql

Then alter your user:

ALTER USER <your username> WITH PASSWORD '<your password>';

You will also want to modify your local configuration to request a password.

Find your hba_file by entering the following in psql:

SHOW hba_file;

This should return something like:/usr/local/var/postgres/pg_hba.conf

You can modify this any way you like. I just use VIM:

vim  /usr/local/var/postgres/pg_hba.conf

At the bottom of the file change the configuration to look like the below: (in VIM press a to enter, then modify your text. Then press esc then type :wq and returnto save your changes and exit).

This will require a password to login for all users.

Now let’s setup our application:

"""Application setup"""

import os

from flask import Flask
from flask_sqlalchemy import SQLAlchemy


def application_setup():
    """creates a flask application and initialises the clothes_shop database"""
    username = os.environ.get("PG_USERNAME", "alex")
    password = os.environ.get("PG_PASSWORD", "test")
    database_name = os.environ.get("DATABASE_NAME", "clothes_shop")

    _app = Flask(__name__)
    _app.config["SQLALCHEMY_DATABASE_URI"] = f"postgresql+pg8000://{username}:{password}@localhost:5432/{database_name}"
    _app.config["SQLALCHEMY_BINDS"] = {f"{database_name}": _app.config["SQLALCHEMY_DATABASE_URI"]}
    _app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

    db = SQLAlchemy(_app)
    db.init_app(_app)

    return _app, db

In this function:

• We get our Postgres username, password and database name from Environment variables.
• We initialise a Flask application.
• We set up our database configuration (URI, Binds and Track Modifications)
• Then initialise our database and application.
• Finally we return the app and database to be used by the main program.

TEST TEST TEST

In this series we will be building the application from a Software Engineering point of view. Any software worth its bytes will be well tested (…not sure on that analogy). This project will be no different.

I like to show you the “less fun” but MOST import side of Software development. Robustly test your code…it will save you a lot of time when it comes to your QA team finding bugs.

Using Pytestwe will be testing our application connection to our database. This is what we call an integration test as we will be testing two of our services interactions within the system.

In your test folder setup a directory called Integration and in that directory add a Python file called test_application_setup.py so that your folder structure looks like:

test/
└── integration/
    └── test_application_setup.py

First we’ll test the happy path inside test_application_setup. add:

"""test_application_setup.py Test the integration on the application setup and interaction with the database"""

from clothes_shop import application_setup


def test_application_setup():
    """Check the database connection is active with correct details"""
    os.environ["DATABASE_NAME"] = "clothes_shop"
    table_name = "clothes_stock"

    app, db = application_setup.application_setup()

    assert app.config["SQLALCHEMY_DATABASE_URI"] == str(db.session.bind.url)
    assert db.session.execute(f"SELECT * from {table_name}")

I like to separate my tests into three parts.

• Setup (test variables, environment variables etc).
• Executing the function to be tested.
• Assertions

In this test we add our database name to an environment variable. Using oswe can set our environment variables. We also set a variable for the table name that we want to do a SQL SELECT on.

We then run the function application_setup which will return the application and database.

Finally, we check that the bind returned from the database configuration is the same as the application configuration. Then most importantly we test that our connection returns a result from the database.

We do this simply from doing a SQLSELECT statement from the database we created in Episode 2. If you remember we inserted some data to a table called clothes_stock. This data should be returned. If not a ProgrammingError will be returned from SQLAlchemy.

Bear in mind for this test we only care about a connection being made. Not the content of the data being returned.

Now let’s do some unhappy paths!

Never just test the path that returns the result you want out of the system. Software Engineers need to think of results that we should not expect, as well as edge case scenarios.

For this project we would not want any incorrect password being able to access privileges of the superuser. The next test we want to make is to test for this:

"""test_application_setup.py Test the integration on the application setup and interaction with the database"""

import os

import pytest
from sqlalchemy.exc import ProgrammingError

from clothes_shop import application_setup


def test_application_setup_incorrect_password():
    """Check the database connection does not allow incorrect passwords"""
    os.environ["PG_USERNAME"] = "alex"
    os.environ["PG_PASSWORD"] = "incorrect_password"
    os.environ["DATABASE_NAME"] = "clothes_shop"
    table_name = "clothes_stock"

    app, db = application_setup.application_setup()

    assert app.config["SQLALCHEMY_DATABASE_URI"] == str(db.session.bind.url)
    with pytest.raises(ProgrammingError):
        assert db.session.execute(f"SELECT * from {table_name}")

In the above case we have set an incorrect password. Then using Pytest.raises we can catch that using that password will raise a ProgrammingError

Finally, we also want to test that an invalid user will not have access to our database:

"""test_application_setup.py Test the integration on the application setup and interaction with the database"""

import os

import pytest
from sqlalchemy.exc import ProgrammingError

from clothes_shop import application_setup

def test_application_setup_invalid_user():
    """Check the database connection does not allow an invalid username"""
    os.environ["PG_USERNAME"] = "invalid_username"
    app, db = application_setup.application_setup()

    assert app.config["SQLALCHEMY_DATABASE_URI"] == str(db.session.bind.url)
    with pytest.raises(ProgrammingError):
        assert db.session.execute("SELECT * from clothes_stock")

In this instance we set a random username. We use the same technique for catching the error as the previous test as this also should raise a ProgrammingError

The full test suite should look like:

"""Test the integration on the application setup and interaction with the database"""

import os

import pytest
from sqlalchemy.exc import ProgrammingError

from clothes_shop import application_setup


def test_application_setup():
    """Check the database connection is active with correct details"""
    os.environ["DATABASE_NAME"] = "clothes_shop"
    table_name = "clothes_stock"

    app, db = application_setup.application_setup()

    assert app.config["SQLALCHEMY_DATABASE_URI"] == str(db.session.bind.url)
    assert db.session.execute(f"SELECT * from {table_name}")


def test_application_setup_incorrect_password():
    """Check the database connection does not allow incorrect passwords"""
    os.environ["PG_USERNAME"] = "alex"
    os.environ["PG_PASSWORD"] = "incorrect_password"
    os.environ["DATABASE_NAME"] = "clothes_shop"
    table_name = "clothes_stock"

    app, db = application_setup.application_setup()

    assert app.config["SQLALCHEMY_DATABASE_URI"] == str(db.session.bind.url)
    with pytest.raises(ProgrammingError):
        assert db.session.execute(f"SELECT * from {table_name}")


def test_application_setup_invalid_user():
    """Check the database connection does not allow an invalid username"""
    os.environ["PG_USERNAME"] = "invalid_username"
    os.environ["DATABASE_NAME"] = "clothes_shop"
    table_name = "clothes_stock"

    app, db = application_setup.application_setup()

    assert app.config["SQLALCHEMY_DATABASE_URI"] == str(db.session.bind.url)
    with pytest.raises(ProgrammingError):
        assert db.session.execute(f"SELECT * from {table_name}")

Then you can run these tests using the following command:

pipenv run pytest

You should see three passing tests!

I hope you found this helpful. The videos for Episodes 3 and 4 will be out shortly. If you have any questions feel free to comment, and I’ll catch you on the next one.