Building a data pipeline: from python objects to a database

There are many ways to store data. There are text files, CSVs (Comma Separated Values), excel tables and so on. They all have their pros and cons. But in this blog post, I am going to talk about one server-based database: Postgres.

I will walk you through the installation of a working environment and a complete example of how one can go from python objects to a database.

What are the needs ?

A library that can convert objects to a data structure understandable by databases, this is called an ORM (Object Relational Mapper). For this we will use the python library SQLAlchemy. It does more than just translating objects from python to databases' data structures, it abstracts away many low level concepts such as connection, querying, and it offers several ways to interact with databases.

Using SQLAlchemy efficiently leads to more readable and easy-to-maintain code. A downside though, I think, is, it is not advisable to use it when we want to learn about SQL. Even though it offers the possibility to work with raw SQL queries.

A database where we'll store our data for later ease of access and querying. Rather than installing the full fledge PostgreSQl, we are going to use a dockerized version of it. Docker makes software installation very easy and takes away all the pain that comes often with installing softwares.

All you need is to have it installed on your machine. For installation guide, go here. And to get the docker, this medium post does a good job.

A python Data base API for communicating with PostgreSQl. You can find a list of options here. For this post, we are going to use psycopg2. SQLAlchemy needs this to properly communicate with the Postgres database.
[Optional] A virtual env which will contain only the necessary packages and where we'll do all the programming work.

PostgreSQL

SQLAlchemy ORM core concepts

At its core, there are Engines. As mentioned on the official website, it is the starting point for any SQLAlchemy application. It contains all the necessary stuffs to interact with a DBAPI (for those interested, you can find here, the full operations for the Python DBAPI) by making them accessible anytime there is a need.

Another core concept for the SQLAlchemy ORM is Session. It is very a convenient technical solution that makes the data persistence easy enough to avoid the user manually reporting all the changes to the created objects. All the queries are dealt with through sessions. For that purpose, it needs to have connection established with the database all the time, and whenever you want to affect changes to existing data or to evaluate queries, a simple notification sent is enough (through an implementation of the DBAPI commit or rollback). It thus works in common with the Engine.

The final concept I would like to talk about before getting to the practice is Table. Creating tables via the SQLAlchemy ORM is as easy as creating classes. The only thing to do is to inherit from a base class which notifies SQLAlchemy that the class should be treated as the definition of a database table.

SQLAlchemy logo

A complete walk through an example

The example project will consist of storing some information about football players in a Postgres database using the SQLAlchemy ORM. To do this I created a project with the following structure:

Project folder structure

a base file where are defined the configurations: Engine, Session and models base class.
a model file where is defined the table for players.
a class containing useful queries for our needs.
a fake player generator file using a very useful library: Faker.

The first thing to do is to define the configurations to use in order to communicate with the database. Here is the code which does that:

"""This code creates: * a SQLAlchemy Engine that will interact with our dockerized PostgreSQL database, * a SQLAlchemy ORM session factory bound to this engine, * a base class for our classes definitions. """ from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker POSTGRES_USER = 'postgres' POSTGRES_PASSWORD = 'docker' POSTGRES_HOST = 'localhost' POSTGRES_PORT = 5432 db_name = 'postgres' DATABASE_URI = f'postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}' \ f'@{POSTGRES_HOST}:{POSTGRES_PORT}/' # create the postgres database engine data_base = create_engine(f'{DATABASE_URI}{db_name}') # make a session binding to the database in a global scope Session = sessionmaker(data_base) # Base class for all the models. Base = declarative_base()

After doing that, we can now define the table associated to players in a declarative way.

"""Create players class and define its characteristics""" from data_engineering.database.base import Base from sqlalchemy import String, Integer, Column, Boolean class Player(Base): __tablename__ = 'players' id = Column(Integer, primary_key=True) name = Column('name', String) nationality = Column('nationality',String) club = Column('club', String) # current if still active, last club if retired total_scored_goals = Column('total_scored_goals', Integer) total_personal_trophies = Column('total_personal_goals', Integer) active = Column('active', Boolean) def __init__(self, name, nationality, club, total_scored_goals, total_personal_trophies, active): """Initialize without the id component as it is taken care of automatically.""" self.name = name self.nationality = nationality self.club = club self.total_scored_goals = total_scored_goals self.total_personal_trophies = total_personal_trophies self.active = active def __repr__(self): """Prettify when we print a player""" if self.active: to_print = f"" else: to_print = f"" return to_print

Now we can define all the operations we want to handle through SQLAlchemy on the database (currently on the players table). This is a class containing some of the operations we can do.

"""Class defining some queries for the database via SQLAlchemy""" from typing import List from data_engineering.database.base import Base, Session, data_base from data_engineering.database.players import Player class DataBase: def __init__(self): # create a session and register players table self.session = Session() Base.metadata.create_all(data_base) def drop_all(self): self.session.query(Player).delete() self.session.commit() def add_new_player(self, player: Player): self.session.add(player) self.session.commit() def add_players(self, players: List[Player]): for player in players: self.session.add(player) self.session.commit() def read_all(self): return self.session.query(Player).all() def delete_entry(self, player): self.session.query(Player).filter(Player == player).delete() self.session.commit() def get_players_from_club(self, club: str = 'Barcelona'): players = self.session.query(Player).filter(Player.club == club).all() return players def delete_players_from_club(self, club: str = 'PSG'): self.session.query(Player).filter(Player.club == club).delete() self.session.commit() def get_top_scorers(self): top_scorers = self.session.query(Player).filter(Player.total_scored_goals >= 600).all() return top_scorers def close(self): self.session.close()

All this done, for tests purpose and to be able to store many players still avoiding the tedious work to create all of them by hand, we need a data generator.

from faker import Faker import random COUNTRIES = ['Argentina', 'Portugal', 'France', 'Brazil', 'England', 'Belgium', 'Italy', 'Spain', 'Germany', 'Ivory coast', 'Guinea', 'Senegal', 'Ghana', 'Nigeria', 'Cameroun', 'Egypt', 'Morocco', 'Algeria', 'USA', 'Croatia', 'Sweden', 'Columbia', 'Netherlands', 'Danemark', 'Ireland', 'Switzerland'] CLUBS = ['Barcelona', 'Real Madrid', 'Juventus', 'Liverpool', 'Ajax', 'Totenham', 'PSG', 'Lyon', 'Napoli', 'Olympique de Marseille', 'Valence FC', 'Manchester United', 'Manchester City', 'Chealsea', 'Athlético Madrid', 'Bayern de Munich', 'Dortmund', 'Arsenal', 'AC Milan', 'Roma', 'Inter', 'Monaco', 'Sevilla'] fake = Faker() def generate_fake_player(): d = dict() d['name'] = lambda: fake.name_male() d['nationality'] = lambda: random.choice(COUNTRIES) d['club'] = lambda: random.choice(CLUBS) d['total_scored_goals'] = lambda: random.randint(0, 700) d['total_personal_trophies'] = lambda: random.randint(0, 20) d['active'] = lambda: random.choice([False, True]) return {k: d[k]() for k in d.keys()}

It is now very easy to test all these. Of course, for this to work, the Postgres should be accessible.

"""Test data base module""" import unittest from data_engineering.database.fake_players import generate_fake_player from data_engineering.database.players import Player from data_engineering.database.postgres_db import DataBase class TestDB(unittest.TestCase): def setUp(self) -> None: """Instantiate a class to work with""" self.postgres = DataBase() def tearDown(self) -> None: """empty table and close connection""" self.postgres.drop_all() self.postgres.close() def test_insert_one_player(self): """Insert one player""" messi = Player(name='Lionel Messi', nationality='Argentina', club='Barcelona', total_scored_goals=686, total_personal_trophies=15, active=True) self.postgres.add_new_player(messi) db_content = [player for player in self.postgres.read_all()] self.assertEqual(1, len(db_content)) self.assertEqual(messi, db_content[0]) def test_insert_more_than_one_player(self): # create players messi = Player(name='Lionel Messi', nationality='Argentina', club='Barcelona', total_scored_goals=686, total_personal_trophies=15, active=True) cristiano = Player(name='Cristiano Ronaldo', nationality='Portugal', club='Juventus', total_scored_goals=680, total_personal_trophies=18, active=True) players = [messi, cristiano] NB = 1000 fake_players = [generate_fake_player() for _ in range(NB)] players.extend([Player(**player) for player in fake_players]) # add them all self.postgres.add_players(players) db_content = [player for player in self.postgres.read_all()] self.assertEqual(NB + 2, len(db_content)) self.assertEqual(messi, db_content[0]) self.assertEqual(cristiano, db_content[1]) # get players from Barcelona: should contain at least Messi barca_players = [b_p for b_p in self.postgres.get_players_from_club('Barcelona')] self.assertGreaterEqual(len(barca_players), 1) self.assertIn(messi, barca_players) # get top scorers: should contain at least Messi and Cristiano top_scorers = [t_p for t_p in self.postgres.get_top_scorers()] self.assertGreaterEqual(len(top_scorers), 2) self.assertIn(messi, top_scorers) self.assertIn(cristiano, top_scorers)

Summary

In this blog, we looked at some concepts of ORM and dive into a practical session using the python library SQLAlchemy to interact with a Postgres database to manipulate a model of football players. We saw how powerful the SQLAlchemy ORM is and how clean it makes the code, avoiding all the intricacies of raw SQL queries.

Now that we have some data available in a database, the next step is to build a data flow around this using a tool such as apache beam.

I really hope you enjoyed this, don't hesitate to send comments.

Note: All the code is available on my Github here.

Building a data pipeline: from python objects to a database

juin 10, 2019 | Tags: database data engineering postgresql data science

What are the needs ?

SQLAlchemy ORM core concepts

A complete walk through an example

Summary