Building Dynamic SQL Queries in Python with Pypika

Pypika is a Python library that provides an easy interface for building SQL queries. It aims to simplify the process of creating dynamic SQL queries in Python.

Pypika provides a rich set of classes and methods that allow us to construct SQL queries in a hierarchical, chainable way, making it easy to add and remove clauses, conditions, and expressions.

Consider we have a database like below:

Here we have three tables, core_product, core_category, and core_product_category.

The “core_product” table has id, name, price, and discount columns.

The “core_category” table has an id and name. Finally the “core_product_category” connect those two tables it has id, product_id, and category_id.

I will give examples based on this schema.

Let’s look at the initially basic example:

from pypika import Query

q = Query.from_('core_product').select("id", "name")

print(q.get_sql())

#SELECT "id","name" FROM "core_product"

It is a very simple example that we used table names and columns as a string but for complex queries, we can use pypika’s table class. Columns can be referenced as attributes in instances of pypika.Table. Now step by step we will discuss more advanced and dynamic queries.

Okay, let’s create our Table classes.

tables.py

from typing import Optional

from pypika import Table, Query
from datetime import datetime


class TypedTable(Table):
    __table__ = ""

    def __init__(
            self,
            name: Optional[str] = None,
            schema: Optional[str] = None,
            alias: Optional[str] = None,
            query_cls: Optional[Query] = None,
    ) -> None:
        if name is None:
            if self.__table__:
                name = self.__table__
            else:
                name = self.__class__.__name__

        super().__init__(name, schema, alias, query_cls)


class Product(TypedTable):
    __table__ = 'core_product'

    id: int
    name: str
    price: int


class Category(TypedTable):
    __table__ = 'core_category'

    id: int
    name: str


class ProductCategory(TypedTable):
    __table__ = 'core_product_category'

    id: int
    product_id: int
    category_id: int

After creating our table classes, now are able to use them in our queries. Let’s write the same query with a new style.

from pypika import Query
from tables import Product

product_tb = Product()

q = Query.from_(product_tb).select(product_tb.id, product_tb.name)

print(q.get_sql())

# SELECT "id","name" FROM "core_product"

Output is the same but we are able to take advantage of autocomplete🥳. It is cool, right?

Usage of the where clause.

from pypika import Query
from tables import Product

product_tb = Product()

q = Query.from_(product_tb).select(product_tb.id, product_tb.name).where(product_tb.price > 10)

print(q.get_sql())
# SELECT "id","name" FROM "core_product" WHERE "price">10

Joins

For joining tables we can use inner_join and on methods.

from pypika import Query
from tables import Product, ProductCategory, Category

product_tb = Product()
category_tb = Category()
product_category_tb = ProductCategory()

q = Query.from_(
    product_tb
).inner_join(
    product_category_tb
).on(
    product_tb.id == product_category_tb.product_id
).select(
    product_tb.id,
    product_tb.name,
    product_category_tb.category_id
).where(
    product_tb.price > 150
)

print(q.get_sql())

"""
SELECT 
  "core_product"."id", 
  "core_product"."name", 
  "core_product_category"."category_id" 
FROM 
  "core_product" 
  JOIN "core_product_category" ON "core_product"."id" = "core_product_category"."product_id" 
WHERE 
  "core_product"."price" > 150

"""

Group by

Here we used the “groupby” method. Also, we used the alias.

from pypika import Query, functions as fn
from tables import Product, ProductCategory, Category

product_tb = Product()
category_tb = Category()
product_category_tb = ProductCategory()

q = Query.from_(
    product_tb
).select(
    category_tb.name,
    (fn.Count(product_tb.id)).as_('product_count')
).inner_join(
    product_category_tb
).on(
    product_tb.id == product_category_tb.product_id
).inner_join(
    category_tb
).on(
    product_category_tb.category_id == category_tb.id
).where(
    product_tb.price > 150
).groupby(category_tb.name)

print(q.get_sql())

"""
SELECT 
  "core_category"."name", 
  COUNT("core_product"."id") "product_count" 
FROM 
  "core_product" 
  JOIN "core_product_category" ON "core_product"."id" = "core_product_category"."product_id" 
  JOIN "core_category" ON "core_product_category"."category_id" = "core_category"."id" 
WHERE 
  "core_product"."price" > 150 
GROUP BY 
  "core_category"."name"

"""

Thanks for reading. I hope you enjoyed it ❤. If you found the article useful don’t forget to clap and follow me.

This is my #12/52 story in 2023, I’m on a challenge to write 52 stories in 2023.

Keep Learning..!

Some other related articles you should go check out!

Methods vs Functions in Python

Python Imports: The Pitfalls of Using ‘*’

How to add a custom field in list_display[Django]