DEV Community: HAP

My Win

HAP — Tue, 08 Aug 2023 17:38:06 +0000

A couple of years ago when I worked at Red Hat, I had filed for a patent as an inventor. You know, they say that the wheels of justice grind slowly. The wheels of patent applications also grind slowly.

But I do have a patent for Red Hat as an inventor!

So, I'm proud of that.

Record Insert with Relational Validation in One SQL Statement

HAP — Thu, 16 Mar 2023 17:25:32 +0000

With Django implementation.

Hello, folks! I'm back with another tidbit regarding extending Django to make it more useful. This time it's regarding objects that are created in a transaction. This can be problematic in Django as it sets foreign key constraints to DEFERRED INITIALLY DEFERRED. "Why?" I hear you ask. Because when constraints are deferred, they are not evaluated until the transaction is committed. This can lead to issues like exceptions not being thrown when they should (especially in test-harness situations like pytest, tox and the like).

So let's say you have a situation where you have models defined similar to these:

class Company(models.Model):
    id = models.BigAutoField(primary_key=True)
    name = models.TextField(null=False)
    ...

class Person(models.Model):
    id = models.BigAutoField(primary_key=True)
    surname = models.TextField(null=False)
    forename = models.TextField()
    midname = models.TextField()
    ...

class Employee(models.Model):
    id = models.BigAutoField(primary_key=True)
    company = models.ForeignKey("Company", null=False, on_delete=models.CASCADE)
    person = models.ForeignKey("Person", null=False, on_delete=models.CASCADE)
    start_date = models.DateTimeField(null=False)
    end_date = models.DatetimeField(null=False)
    ...

Your standard-ish employee example, right? So let's say we have a company record:

{"id": 10, "name": Saskatoon Widgets, Inc."}

And a person record:

{"id": 45651, "surname": "Bond", "forename": "James", "midname": "Pootwaddle"}

So now, he got hired and we want to make him an employee. What if this is an API call? How will we know if the foreign key values are correct? I know you are all like Horshack and yelling "Serializers"! But, what if you're not validating using serializers? (It could happen!) Just follow along with me for a bit longer.

In autocommit mode (or if you're using short-lived transactions and you're not kicking off any other actions that would be inside of the transaction) the database driver would return an IntegrityError exception that could be handled.

def create_employee(emp_record: Dict):
    try:
        emp = Employee.objects.create(**emp_record)
    except IntegrityError as e:
        return Http422Response(str(e))
    return HttpResponse(emp)

...

rec = {"company": 10, "person": -1, ...}
create_employee(rec) # You get the 422 in autocommit or transaction execution

But what if you use the newly created emp object with other functions while within the transaction?

def create_employee(emp_record: Dict):
    try:
        emp = Employee.objects.create(**emp_record)
        queue_new_hire_actions(emp) # Uh-oh!
    except IntegrityError as e:
        return Http422Response(str(e))
    return HttpResponse(emp)

...

rec = {"company": 10, "person": -1, ...}
create_employee(rec)

So you'd get some sort of exception here which may or may not get handled in the defined exception block above. But you also now have a queued record that will fail later as well.

So if you use serializers for validation or if you make your own validation functions, you run the risk of hitting the DB with extra queries. But in a situation where there's a lot of activity against an endpoint, you end up throwing a lot of extra query traffic against the database.

There is an alternative. Use a CTE to return the inserted record and use that record data to check the related tables. Something like this:

with new_rec as (
insert into employee (company_id, person_id, start_date)
values (10, -1, now())
returning *
)
select nr.id,
       nr.start_date, 
       cmp.id as "company_id",
       prs.id as "person_id"
  from new_rec as nr
  left
  join company as cmp
    on cmp.id = nr.company_id
  left
  join person as prs
    on prs.id = nr.person_id;

So this would return a record that would match the class attributes for the Employee model. But now we can check for bad data immediately because, in this case, the returned "person_id" would be null which we could check immediately after create.

This can be extended to fetching immediately related records as well (that is, not walking down all relations).

with new_rec as (
insert into employee (company_id, person_id, start_date)
values (10, -1, now())
returning *
)
select nr.id,
       nr.start_date, 
       cmp.id as "company_id",
       row_to_json(cmp) as "company",
       prs.id as "person_id",
       row_to_json(prs) as "person"
  from new_rec as nr
  left
  join company as cmp
    on cmp.id = nr.company_id
  left
  join person as prs
    on prs.id = nr.person_id;

So now you get the keys and the table record (as json) on query return. If missing, the key and the record will be null.

This type of query could more easily be done with SQLAlchemy, but we're talking about Django here. So we have to use some raw SQL building and (eventually) return a model instance.

Here's what I did for a Django implementation:

The input to the whole thing will be a model class for the target table and a dict for the data to be inserted.

These are the imports I've used along with some module-level globals. The ALIASES list is used to grab table aliases that will be consistently be used in the statement build.

import os
from datetime import datetime, timezone
from enum import Enum
from typing import List

from django.db import IntegrityError, connection, models

LETTERS = "abcdefghijklmnopqrstuvwxyz"
ALIASES = [f"{'_' * i}{letter}" for i in range(1, 3) for letter in LETTERS]

So, first, I need to ensure that the model defaults are applied to the record.

def apply_model_defaults(model: models.Model, record: dict) -> dict:
    now = datetime.now(tz=timezone.utc)
    full_record = record.copy()

    for field in model._meta.concrete_fields:
        if field.primary_key:
            continue

        if isinstance(field, models.ForeignKey):
            fname = field.get_attname_column()[-1]
        else:
            fname = field.name

        if getattr(field, "auto_now_add", False) or getattr(
            field, "auto_now", False
        ):
            default = now
        else:
            default = field.default

        if default != models.NOT_PROVIDED:
            if fname not in record:
                if callable(default):
                    full_record[fname] = default()
                elif isinstance(default, Enum):
                    full_record[fname] = str(default.value)
                else:
                    full_record[fname] = default

        if not field.null and full_record.get(fname) is None:
            raise ValueError(f"{model.__name__}.{fname} cannot be None.")

    return full_record

And I also need a function to resolve a model reference in the record to the model's primary key value.

def fk_or_model_pk(fkeys: List[models.Field], record: dict) -> dict:
    for f in fkeys:
        if f.name in record:
            if isinstance(record[f.name], models.Model):
                record[f.name] = getattr(
                    record[f.name], f.target_field.name, None
                )
    return record

Next, I need to be able to generate the insert statement.

def build_insert(
    model: models.Model, fkeys: List[models.Field], record: dict
) -> str:
    """Build the base insert."""
    field_to_target_col = {f.name: f.get_attname_column()[-1] for f in fkeys}
    return """
insert into {table} ({column_list})
values ({data_list})
returning *
    """.format(
        table=model._meta.db_table,
        column_list=", ".join(
            field_to_target_col.get(col, col) for col in record
        ),
        data_list=", ".join(f"%({col})s" for col in record),
    )

I need an select statement generator that will wrap the insert in a CTE, then select all of the foreign key table info that I need.

def build_insert_validated_select(
    model: models.Model, record: dict, fetch_related: bool = False
) -> str:
    """Wrap the base insert in a CTE for related table data verification.

    Immediate relations of the model can be fetched as json. (fetch_related)
    """
    remote_fields = []
    # Get the foreign key fields and build a dict holding
    # The necessary parts to generate the select columns 
    # and left-joins
    fk_fields = [
        {
            "remote_alias": ALIASES[fnum],
            "join_local_key": f"_nr.{f.get_attname_column()[-1]}",
            "local_key": f.get_attname_column()[-1],
            "remote_table": f.related_model._meta.db_table,
            "remote_key": f"{ALIASES[fnum]}.{f.target_field.name}",
            "remote_row": f"row_to_json({ALIASES[fnum]}) as {f.related_model._meta.db_table}_rec",  # noqa E501
            "_field": f,
        }
        for fnum, f in enumerate(model._meta.concrete_fields)
        if isinstance(f, models.ForeignKey)
    ]
    # Local fields are all defined fields that are
    # not ForeignKey instances.
    local_fields = [
        f"_nr.{f.name}"
        for f in model._meta.concrete_fields
        if not isinstance(f, models.ForeignKey)
    ]
    # Build a list of remote fields with aliases from the 
    # fk_fields list
    remote_fields = [
        f"{fk['remote_key']} as {fk['local_key']}" for fk in fk_fields
    ]
    # If fetch related, include the row_to_json() calls
    if fetch_related:
        remote_fields.extend(fk["remote_row"] for fk in fk_fields)
    # Build the left joins from the fk_fields
    if fk_fields:
        left_joins = f"{os.linesep}  ".join(
            "left join {remote_table} as {remote_alias} "
            "on {remote_key} = {join_local_key}".format(**fk)
            for fk in fk_fields
        )
    # Some pretty print formatting
    sep_indent = f",{os.linesep}       "
    select_cols = sep_indent.join(local_fields + remote_fields)
    insert_sql = build_insert(model, [f["_field"] for f in fk_fields], record)
    return f"""
with new_rec as (
{insert_sql}
)
select {select_cols}
  from new_rec as _nr
  {left_joins}
;
    """

And finally, I need the main function call that will build and execute the statement and perform the post-insert data check.

def validated_create(
    model: models.Model, record: dict, fetch_related: bool = False
) -> models.Model:
    """Create record for model and verify existence of related data."""
    # Apply the model defaults to the data record dict
    record = apply_model_defaults(model, record)
    fk_fields = [
        f
        for f in model._meta.concrete_fields
        if isinstance(f, models.ForeignKey)
    ]
    # Resolve model references to primary key values
    record = fk_or_model_pk(fk_fields, record)

    # Build the statement
    val_ins_sel_sql = build_insert_validated_select(
        model, record, fetch_related=fetch_related
    )
    val_rec = None

    # Execute the statement and fetch the result 
    # as a dict
    with connection.cursor() as cur:
        cur.execute(val_ins_sel_sql, record)
        val_rec = dict(zip([d[0] for d in cur.description], cur.fetchone()))

    # Validate foreign key existence for any foreign keys in the input.
    # This is necessary in case the SQL is executed as part of a transaction
    # which will result in deferred constraint validation as django sets
    # foreign key constraints as deferred.
    for f in fk_fields:
        # Resolve key names for input and output
        # foreign key references
        if f.name in record:
            fkname = f.name
        elif f.get_attname_column()[-1] in record:
            fkname = f.get_attname_column()[-1]
        else:
            continue
        vfkname = f.get_attname_column()[-1]

        # Check to see if the values of the foreign
        # keys are different. If so, throw exception.
        # This is how the integrity checking is done
        # during a transaction.
        if val_rec[vfkname] != record[fkname]:
            related_table = f"{f.related_model._meta.db_table}"
            msg = "is not present in table"
            raise IntegrityError(
                f"Key ({fkname})=({record[fkname]}) {msg} {related_table}"
            )

        # instantiate related model reference if it exists and is not None
        remote_table_name = f.related_model._meta.db_table
        remote_table_ref = f"{remote_table_name}_rec"
        if remote_table_ref in val_rec:
            remote_table_rec = val_rec.pop(remote_table_ref)
            if not remote_table_rec:
                val_rec[f.name] = f.related_model(**remote_table_rec)

    # return a target model instance using the 
    # fetched data
    return model(**val_rec)

So probing the model class structures, I can now build sql that should work for any properly defined Django ORM model class.

The point behind all of this is to validate related record existence without sending a query per model.

Message to those in the United States: Please go vote!

HAP — Mon, 07 Nov 2022 22:39:52 +0000

Public Service Announcement

I really don't want to get bogged down in a political thread here. But I do want to simply urge US citizens who are eligible to vote, to make sure you're registered and please go vote in the November 8 midterm election.

There are so many messages floating around and the political ads are enough to emotionally and mentally beat you up pretty good. But to my fellow citizens, I say: "The United States is still a representative democracy as of this post. Please exercise your civic duty and vote!"

Try have at least a basic understanding of issues that matter to you. Please do, at least, a cursory look into voting histories of your candidates to see if their actions match their rhetoric.

You may not think your vote matters, but it does.

Plea to Moderators

I hope I have not offended anyone on this site. But this is important. I will understand the removal of this post if it violates site guidelines.

Python Throws a Bone to Other Developers

HAP — Tue, 01 Nov 2022 16:50:53 +0000

This may be a bit late to the party, but I recently found out that, starting with Python 3.8, You can assign variables within expressions. This was something I really disliked about Python way back in the 2.5 days when I switched over from C.

So, now instead of a code block like this:

if len(data) > 0:
    process(data)
print(f"Processed {len(data)} bytes!")

You can combine variable assignment and the comparison in one line of code!

if (dat_len := len(data)) > 0:
    process(data)
print(f"Processed {dat_len} bytes!")

It's all thanks to the addition of the := operator.

For those of you who started programming in Python, this may not be such a big deal. But for those of us who come to Python from other languages, this is a real benefit, IMHO, as it helps the transition of programming in a "Pythonic" way that is also similar to the programming patterns from other languages.

See the docs here.

A neat Python buffer read iterable.

HAP — Fri, 02 Sep 2022 14:57:20 +0000

I've recently discovered a Python "nifty" regarding iter(). We've seen this form of iterable file read a million times:

with open('eek.txt', 'rt') as f:
    for line in f:
        # Some super-cool stuff!

But!! Did you know you can read arbitrary-length buffers in a similar form? Use the iter(object, sentinel) form of the call and you can do this:

with open('eek.txt', 'rt') as f:
    # Use whatever chunk size you want
    for buff in iter(lambda: f.read(1024), ''):
        # Super-cool stuff using a 1K buffer!!

The use of iter(read_chunk, '') is basically saying: "Call read_chunk until you get an empty string." This could be done with binary files as well. The mode would be 'rb' and the sentinel would be b''.

Neato, eh?

iter function documentation

Understanding a Data Model

HAP — Wed, 20 Jul 2022 15:31:55 +0000

Or "How I learned to stop forgetting things and use comments."

One of the nifties of PostgreSQL is support for comments on objects. You can comment on tables and columns and so many more objects.

Let's try an example:

create table my_cool_table (
    id serial primary key,
    cool_rating numeric(5,2) not null default 0.0::numeric(5,2),
    label text not null,
    data jsonb not null default '{}'::jsonb
)
;

comment on table my_cool_table is 'Cool Ratings';
comment on column my_cool_table.id is 'Primary Key';
comment on column my_cool_table.cool_rating is 'Just how cool this is.';
comment on column my_cool_table.label is 'Label associated with data';
comment on column my_cool_table.data is 'The cool stuff';

OK, so now what?

Well, you can now display the comments when getting information. For instance, using psql you can \dt+ and see the table comment.

postgres=# \dt+ my_cool_table
                                  List of relations
 Schema |     Name      | Type  |  Owner   | Persistence |    Size    | Description  
--------+---------------+-------+----------+-------------+------------+--------------
 public | my_cool_table | table | postgres | permanent   | 8192 bytes | Cool Ratings
(1 row)

You can get the column comments using \d+ my_cool_table

postgres=# \d+ my_cool_table
                                                             Table "public.my_cool_table"
   Column    |     Type     | Collation | Nullable |                  Default                  | Storage  | Stats target |        Description         
-------------+--------------+-----------+----------+-------------------------------------------+----------+--------------+----------------------------
 id          | integer      |           | not null | nextval('my_cool_table_id_seq'::regclass) | plain    |              | Primary Key
 cool_rating | numeric(5,2) |           | not null | 0.0::numeric(5,2)                         | main     |              | Just how cool this is.
 label       | text         |           | not null |                                           | extended |              | Label associated with data
 data        | jsonb        |           | not null | '{}'::jsonb                               | extended |              | The cool stuff
Indexes:
    "my_cool_table_pkey" PRIMARY KEY, btree (id)
Access method: heap

There are other programs that can access the comments. dBeaver, for instance:

There are also some documentation programs that will crawl your DB and output various format files that can include comments.

SQLAlchemy ORM should support comments if you're using it to create database objects.

Use of the PostgreSQL comments can really help new developers but be a handy reference whenever changes are necessary.

See the PostgreSQL COMMENT documentation to see all of the objects on which comments are supported.

Python, Celery, Kubernetes, and Memory

HAP — Tue, 12 Apr 2022 16:26:26 +0000

I've been thinking about Python and Kubernetes and long-lived pods (such as celery workers); and I have some questions and thoughts.

When you use memory-intensive modules like pandas, would it make more sense to have the "worker" simply be a listener and fork a process (passing the environment, of course) to do the actual processing using memory-intensive modules? The thought process here is that, by launching a subprocess, the memory utilization should go down once that subprocess exits. Therefore memory should be freed and any chance of memory leaks should be circumvented.

Secondly, with kafka and faust, is celery even relevant for high-availability microservice applications?

I would really like to hear some real-world experience with any of these.

A Word About psycopg2 Exceptions

HAP — Wed, 03 Nov 2021 14:50:55 +0000

And the word is "Information."

There is a wealth of information that is captured in the psycopg2 exceptions. These exceptions not only capture a (slightly) reformatted message in the standard args attribute, but also have more attributes that have information from the driver and the cursor and connection classes.

The extra attributes are:

Attribute	Description
cursor	The executing cursor during the exception\|
diag	An object that contains extra information regarding the context and the statement that was running. This may also capture syntax errors in the SQL.
pgcode	The raw PostgreSQL error code
pgerror	The raw PostgreSQL error message\|

So what does that do for you? The cursor attribute is a cursor object. This means that you have access to the last run query as well as the connection. So those can be probed for more information such as the query text and the connection back-end pid. The diag can help with sql statement syntax errors. The pgcode and pgerror can be useful for lower-level analysis, if needed.

Be aware that if you need to query the DB for more information during an exception, you should use a separate connection to get it as an exception state from SQL execution will put the transaction in a state that must be rolled back.

With Django exceptions, you need to access a dunder attribute to get to the information you will need. Django exceptions for the database will most likely only have the args attribute. But it does have the __cause__ attribute which should be the original driver exception instance. And from that instance, you can access the other attributes.

This can prove quite helpful if you need more information quickly without actually trying to access search and parse database log information.

See the this documentation for more information regarding the psycopg2 exception classes representing PostgreSQL errors.

Happy coding and happy exception handling!

Dynamically Create ORM Models for Individual Table Partitions

HAP — Tue, 12 Oct 2021 16:37:26 +0000

Some examples you may find on the internet will describe creating models for partitions by hand. What a pain! This does not allow for dynamic data expansion and utilization. This is a short article regarding a solution for the Django to dynamically create ORM models for individual table partitions. The database backend is PostgreSQL, for those interested.

For those of you who haven't followed my partitioning adventures with django, I do have statically defined models for the partitioned tables, just not for the individual partitions. Currently, all work utilized the interface provided by the partitioned table. But I wanted to see if there was a way to possibly increase efficiency by accessing table partitions directly. I just didn't want to write another SQL builder. But how to do this without writing one?

I first tried to dynamically create a subclass of the partitioned table. Here be dragons! What initially surprised me the most was that it worked in terms of compiling and creating the class object. What harshed my mellow was that dynamically created pointer fields to the superclass were created and would cause their own havoc. So what to do? Now I had to figure out how to create the necessary meta internal class and field definitions with all of the correct (or at least application-required) options set properly. With all of the myriad options for fields, that is a daunting task.

Or so I thought. Taking another deep dive through all of the settings in the model class and fields for an application ORM model, I struck gold here:

<model_class>._meta.fields[<some_index>].db_type_parameters()

This returns a dict of all of the parameters that were set for the field (including any derived ones). I tried to just use the dict directly, but you end up with errors because of the derived fields (those keys starting with an underscore, among some others). There were also issues because the to and on_delete are not populated in that dict for ForeignKey fields. So, knowing the minimum number of parameters needed to properly create an ORM model class for a partition of a table, I constructed a set of "allowed" keys to be the minimum to create a model field and I had to probe the field relation to get the to and on_delete values for the ForeignKey fields.

Here's the final form of the suite of functions:

from django.db import transaction
from django.db.models.fields.related import RelatedField

def _create_partition_model(partition_info):
    class _Meta:
        db_table = partition_info.table_name

    model = get_model(partition_info.partitioned_table_name)

    partition_model_name = model.__name__ + _get_partition_model_suffix(partition_instance)

    partition_model_attrs = {
        "Meta": _Meta,
        "__module__": model.__module__,
    }
    allowed_fields = {"null", "primary_key", "default", "editable", "serialize", "db_column", "base_field"}

    for field in model._meta.fields:
        partition_field_attrs = {
            k: v for k, v in field.db_type_parameters(transaction.get_connection()).items() if k in allowed_fields
        }
        if isinstance(field, RelatedField):
            partition_field_attrs["to"] = field.related_model.__name__
            partition_field_attrs["on_delete"] = models.DO_NOTHING
            for rel_obj in field.related_model._meta.related_objects:
                if rel_obj.remote_field == field:
                    partition_field_attrs["on_delete"] = rel_obj.on_delete

        partition_model_attrs[field.name] = type(field)(**partition_field_attrs)

    _model = type(partition_model_name, model.__bases__, partition_model_attrs)

    return _model

Let's break it down:

def _create_partition_model(partition_info):
    class _Meta:
        db_table = partition_info.table_name

    model = get_model(partition_info.partitioned_table_name)

    partition_model_name = model.__name__ + _get_partition_model_suffix(partition_instance)

Since we're working with the same field definition, but a different database table, we need to set the Meta.db_table to the actual table partition name.

The partition_info object is simply an object instance that holds information regarding the target table partition.

Substitute get_model() with whatever means you would use to dynamically get model classes from django. It'll be some form of django.apps.apps.all_models probably.

This assumes that partitions are named as (partitioned_table_name + partition_suffix) In this function, the partition model name (ORM model class name) is derived from the partitioned table class name plus the db table partition name suffix. Substitute _get_partition_model_suffix with whatever means you would use to get the table partition suffix or with whatever name mangling you prefer.

    partition_model_attrs = {
        "Meta": _Meta,
        "__module__": model.__module__,
    }
    allowed_fields = {"null", "primary_key", "default", "editable", "serialize", "db_column", "base_field"}

Now we are creating the dict of class attributes for the new class. The allowed_fields is the base set of fields required (in our app, anyway) to create a field. Adjust these values as needed. You will need base_field for things like an array of another type.

    for field in model._meta.fields:
        partition_field_attrs = {
            k: v for k, v in field.db_type_parameters(transaction.get_connection()).items() if k in allowed_fields
        }
        if isinstance(field, RelatedField):
            partition_field_attrs["to"] = field.related_model.__name__
            partition_field_attrs["on_delete"] = models.DO_NOTHING
            for rel_obj in field.related_model._meta.related_objects:
                if rel_obj.remote_field == field:
                    partition_field_attrs["on_delete"] = rel_obj.on_delete

        partition_model_attrs[field.name] = type(field)(**partition_field_attrs)

Now we're building all of the fields for the new class from the partitioned table's ORM class fields. Note that we are constraining the field parameters to the allowed_fields set.

When we have detected a field type class derived from RelatedField we need to set to and on_delete properly. The to attribute is easy. The on_delete attribute is a bit more difficult. The solution I found that works well is to loop through the field's related objects to find the one relation that points back to the field in question. There may be a better way to do this.

Finally, we add the field to the model attributes using the field attributes we've collected.

    _model = type(partition_model_name, model.__bases__, partition_model_attrs)

    return _model

Finally, we create the model class using type(). It was important here to use the __bases__ from the partitioned table's model. Using the model directly creates the problematic subclass.

The cool part is that django will also register the new model class in its all_models dict due to the metaprogramming going on under the hood.

And that's how I did it. Now we can use the new ORM model class to perform CRUD on the partition directly. And we didn't have to taint the original model to do so.

So now, there is the possibility to process data in multiple threads or processes and access the partitions directly without having to write a bunch of permutations of table partition classes. So long as the structure is consistent between the partitioned table and its partitions, this should work seamlessly.

I hope those projects taking advantage of partitioning in PostgreSQL and using django for the webapp will find this useful. It would really surprise me if there was not some sort of way to also do this with SQLAlchemy.

Handle Default Partition Constraint Collision on New Partition

HAP — Mon, 13 Sep 2021 22:22:11 +0000

Just wanted to drop a quick-ish post on how I've been handling creating a new partition that might have overlapping data in the default partition. We have adopted using a default partition in order to capture as much data as possible during processing without throwing exceptions when a defined partition was not created.

In my team's project, we are using partitioned tables using a date range for the table partitions. We've adopted a naming convention of <table_name>_default for the default partition and <table_name>_YYYY_MM for the monthly partitions. We process a lot of data, so we don't want to do more data manipulation than we have to. Also, this project is a web app, so we have to deal with a lot of concurrent database requests. So, using a lot of temp tablespace was not desirable either. So, here's a breakdown of how I'm handling default partition collisions when creating a new partition.

First: Get all of the information needed to create the partition. This includes the table partition name, the partitioned table name, the partition key column, the partition parameters (in our case, it's default = False, from = <partition-start-date>, to = <partition-end-date>).

Second: Check if any partition key col value within the requested partition range exists in the default partition. If data does exist (which would cause a PostgreSQL error), I alter the from, to parameters adding an arbitrary year delta (such as 100 years) to each of the boundaries. Then I create the partition (using the original name) for the bounds that will not match. Don't forget to save the original values.

Third: Now that I have a partition created, I detach the partition from the partitioned table. Then I copy the overlapping data from the default partition to the detached partition. Once that is complete, I delete that data from the default. I use the original values for the from, to parameters to reattach the partition to the partitioned table.

This method allows me to move the data without creating a potentially big temp table (temp tablespace) or a regular table in the app tablespace that will be deleted later. It also means that I can let PostgreSQL create the table as it sees fit for a partition. Since a partition can be detached and later re-attached with different constraints, this was a feasible option for our application.

Obviously, this will impact performance if it occurs often or with a large amount of data, but the answer to that is to handle potential edge cases in the application code that would see data hit the default partition.

Hopefully this methodology could prove viable to other developers and/or architects.

See the PostgreSQL docs on partitioning.
See the ALTER TABLE documentation for detach and attach partition clauses.

Table Partitioning with django

HAP — Thu, 19 Aug 2021 17:49:43 +0000

"This is my django partitioning solution. There are many like it, but this one is mine."

The impatient can jump to the codey bits.

Background

The axiom can be applied to anything, but "django is great... until it isn't". There is no explicit or implicit support for partitioning in django, but it's not impossible. Join me for my tale of hacking a solution to this problem.

Are you sitting comfortably? Then we'll begin.

I had already made some code to "convert" a table in postgres to a partitioned table. It, too, worked great... until it didn't. I had tried some other modules, but they all had their problems and simply did not work with our app or how we are deploying. Despite our convert code, we still had the issue of being unable to initially create a partitioned table. So I took a deep dive into the django code and eventually found the pieces I needed in the BaseDatabaseSchemaEditor class.

So, I tried to subclass this and figure out how to tell the migration app to use it. But with all of my searching (and a definite time window for completion) it looked like I would have to write my own migration app. This was distasteful because we were already using a tenant module's app for schema migrations and I really did not want to write yet another django app just to change one SQL statement and emit a few deferred ones. So I focused on the methods in BaseDatabaseSchemaEditor to see exactly what happened during create and delete model (table).

In the BaseDatabaseSchemaEditor class, there were two methods I focused on: table_sql for creating a table; and delete_model for dropping a table. Once I had read over these methods and understood what they were doing, I formulated my hack. This required some method overrides, and enable function, a disable function, a list of known partitioned model names, a partition information metaclass, and two support functions.

Python Code

Support Functions

I wanted a function that would be able to get me any model I wanted, globally. It needed to be able to return a model based on the model_name, app_label.model_name or table_name. This function's data would need to be cached and that cache loaded at runtime to ensure that all models were compiled and loaded into django's app/model cache.

_load_db_models

def _load_db_models():
    """Initialize the global dict that will hold the table_name/model_name -> Model map"""
    qualified_only = set()
    for app, _models in django.apps.apps.all_models.items():
        for lmodel_name, model in _models.items():
            qualified_model_name = f"{app}.{lmodel_name}"
            if lmodel_name in DB_MODELS:
                qualified_only.add(lmodel_name)
                del DB_MODELS[lmodel_name]

            DB_MODELS[qualified_model_name] = model
            DB_MODELS[model._meta.db_table] = model
            if lmodel_name not in qualified_only:
                DB_MODELS[lmodel_name] = model

This uses a module-level cache variable named DB_MODELS defined as an empty dict. There is also a module-level lock defined as:

import threading
...
DB_MODELS_LOCK = threading.Lock()

get_model

def get_model(model_or_table_name):
    """Get a model class from the model name or table name"""
    with DB_MODELS_LOCK:
        if not DB_MODELS:
            _load_db_models()
    return DB_MODELS[model_or_table_name.lower()]

Thus if I had an app named eek and an ORM model defined in this app as

from django.db import models
...
class MySuperCoolModel(models.Model):
    class Meta:
        db_table = "my_super_cool_model"
    id = models.UUIDField()
    start_date = models.DateField()
    ...

Then each these calls would all return the MySuperCoolModel class:

get_model("mysupercoolmodel")
get_model("MySuperCoolModel")
get_model("eek.MySuperCoolModel")
get_model("my_super_cool_model")

PartitonedModels

This is simply a list of model names that should be partitioned when created. I called my list PARTITIONED_MODEL_NAMES

Metaclass

This is a ORM model attribute that itself is a class containing information needed to make a partitioned table. It takes this form:

class PartitionInfo:
    partition_type = <str: a partition type such as "RANGE" or "LIST" or "HASH">
    partition_cols = <list[str]: a list of ORM model columns (this needs to match the format of the column as it would appear in the database>

As applied to MySuperCoolModel:

PARTITION_MODEL_NAMES = [
    ...
    "MySuperCoolModel",
]

class MySuperCoolModel(models.Model):
    class PartitionInfo:
        partition_type = "RANGE"
        partition_cols = ["start_date"]
    class Meta:
        db_table = "my_super_cool_model"
    id = models.UUIDField()
    start_date = models.DateField()
    ...

Override Methods

table_sql

I made a new function called p_table_sql (p_ for partitioned). This function will call the original function which I call o_table_sql (o_ for original) to get the initial SQL created as normal. Then I check to see if there is a name match from the model class passed in against the PARTITIONED_MODEL_NAMES list. If there is a match, I use get_model to retrieve the full ORM model class.

If I have a name match and the ORM model has the PartitionInfo metaclass, then I proceed with the new functionality. Otherwise, I return the original sql.

The new functionality consists of adding the PARTITION BY <type> (<cols>) clause to the create table SQL. Also the PRIMARY KEY qualifier is removed from the SQL. The primary key is then added to the deferred SQL list as an ALTER TABLE statement. This is inserted into the deferred_sql list before any FOREIGN KEY statements. If you have needs for tracking information or other post-create sql to be run for the new table, append that sql to the deferred_sql list.

This is because django does not support multiple column primary keys well, if at all. For partitioned models, it is best that any column that you would consider a primary key be created explicitly and not be a automatic incrementing field. UUID type is a good candidate for this.

from django.db.models import ForeignKey
...
def _count_fk_fields(model):
    num_fk = 0
    for f in model._meta.fields:
        num_fk += int(isinstance(f, ForeignKey))
    return num_fk


def p_table_sql(self, model):
    # Use default model class for the original django SQL generation
    sql, params = self.o_table_sql(model)

    # Based on model name match, get the defined model from the app
    # from the django migration processing
    if model.__name__ in PARTITIONED_MODEL_NAMES:
        pmodel = get_model(model.__name__)
    else:
        pmodel = None

    # If there was a partition name match and the class has the required attribute,
    # use this information to add the partition clause to the create table sql
    # Otherwise, return the original sql and params
    if pmodel is not None and hasattr(pmodel, "PartitionInfo"):
        LOG.info(f"Creating PARTITIONED TABLE {pmodel._meta.db_table}")
        partition_cols = pmodel.PartitionInfo.partition_cols
        sparams = {
            "partition_type": pmodel.PartitionInfo.partition_type.upper(),
            "partition_cols": ", ".join(f'"{c}"' for c in partition_cols),
        }

        # The primary key will be overridden here
        # Partitioned tables require that the partition column(s) be part of the primary key
        p_sql = self.sql_partitioned_table % sparams
        sql = sql.replace("PRIMARY KEY", "") + p_sql

        pk_cols = partition_cols[:]
        try:
            mod_pk = pmodel._meta.pk.get_attname()
        except Exception:
            pass
        else:
            pk_cols.append(mod_pk)

        sparams = {
            "table_name": f'"{pmodel._meta.db_table}"',
            "constraint_name": f'"{pmodel._meta.db_table}_pk"',
            "constraint_cols": ", ".join(f'"{c}"' for c in pk_cols),
        }
        pk_constraint = self.sql_partitioned_pk % sparams

        num_fk = _count_fk_fields(pmodel)
        if num_fk:
            self.deferred_sql.insert(-num_fk - 1, pk_constraint)
        else:
            self.deferred_sql.append(pk_constraint)

        # Add any deferred sql statement for post-create to
        # the self.deferred_sql list
    else:
        LOG.info(f"Creating TABLE {model._meta.db_table}")

    return sql, params

delete_model

This function (which I've named p_delete_model) needs to do any pre-processing before the table is dropped. Use the original functionality (which I've reassigned to o_delete_model) to do the actual table drop.

For my implementation, I have a tracking table that lists all of the partitions of any of my partitioned tables. So I want to clean that up.

def p_delete_model(self, model):
    if model.__name__ in PARTITIONED_MODEL_NAMES:
        pmodel = get_model(model.__name__)
    else:
        pmodel = None

    if pmodel is not None and hasattr(pmodel, "PartitionInfo"):
        sparams = {"partitioned_table_name": pmodel._meta.db_table}
        with self.connection.cursor() as cur:
            drop_partitions_sql = cur.mogrify(self.sql_drop_partitions, sparams).decode("utf-8")
        self.execute(drop_partitions_sql)

    self.o_delete_model(model)

Enable Function

This function will set SQL templates for partitioning as well as reassign original functionality to o_ attributes and assign the new functionality to the original attributes.

def set_partitioned_schema_editor(schema_editor):
    """
    Add attributes and override method of given schema_editor to allow partition table sql statements
    to be emitted.
    """
    # Add SQL templates, if not already present
    if not hasattr(schema_editor, "sql_partitioned_table"):
        setattr(schema_editor, "sql_partitioned_table", " PARTITION BY %(partition_type)s (%(partition_cols)s) ")

    if not hasattr(schema_editor, "sql_partitioned_pk"):
        setattr(
            schema_editor,
            "sql_partitioned_pk",
            "ALTER TABLE %(table_name)s ADD CONSTRAINT %(constraint_name)s PRIMARY KEY (%(constraint_cols)s)",
        )

    # Template to drop partitions by using the partition manager trigger function set
    # on the table
    if not hasattr(schema_editor, "sql_drop_partitions"):
        drop_partitions_sql = """
DELETE
  FROM partitioned_tables
 WHERE schema_name = current_schema
   AND partition_of_table_name = %(partitioned_table_name)s
"""
        setattr(schema_editor, "sql_drop_partitions", drop_partitions_sql)

    # Backup original method to emit create table sql and replace with the new method
    if not hasattr(schema_editor, "o_table_sql"):
        setattr(schema_editor, "o_table_sql", schema_editor.table_sql)
        setattr(schema_editor, "table_sql", types.MethodType(p_table_sql, schema_editor))

    if not hasattr(schema_editor, "o_delete_model"):
        setattr(schema_editor, "o_delete_model", schema_editor.delete_model)
        setattr(schema_editor, "delete_model", types.MethodType(p_delete_model, schema_editor))

Disable Function

This function will restore the schema_editor instance back to default.

def unset_partitioned_schema_editor(schema_editor):
    # Delete partition template attributes, if present
    if not hasattr(schema_editor, "sql_partitioned_table"):
        delattr(schema_editor, "sql_partitioned_table")

    if not hasattr(schema_editor, "sql_partitioned_pk"):
        delattr(schema_editor, "sql_partitioned_pk")

    if not hasattr(schema_editor, "sql_drop_partitions"):
        delattr(schema_editor, "sql_drop_partitions")

    # Restore original functionality for create table sql emit
    if not hasattr(schema_editor, "o_table_sql"):
        setattr(schema_editor, "table_sql", schema_editor.o_table_sql)
        delattr(schema_editor, "o_table_sql")

    # Restore original functionality for delete_model method
    if not hasattr(schema_editor, "o_delete_model"):
        setattr(schema_editor, "table_sql", schema_editor.o_delete_model)
        delattr(schema_editor, "o_delete_model")

Migrations

Once all of this is done, I had to utilize it in a migration file. After I dug through the django logic that ran migrations, I could see that (at least with migrate_schemas from django_tenant_schemas) that the schema_editor was created as a context manager and utilized throughout all migration files to be applied from the current state.

This means that I can set and clear the schema editor from within a migration file as discrete RunPython migration operations. The first operation should be to set the schema_editor for partitioning with the reverse action to unset. The last operation should be to unset the schema_editor with the reverse action to set.

Example:

from <your_partition_code_module> import set_partitioned_schema_editor
from <your_partition_code_module> import unset_partitioned_schema_editor

def set_partition_mode(apps, schema_editor):
    set_partitioned_schema_editor(schema_editor)


def unset_partition_mode(apps, schema_editor):
    unset_partitioned_schema_editor(schema_editor)


class Migration(migrations.Migration):

    dependencies = []

    operations = [
        migrations.RunPython(code=set_partition_mode, reverse_code=unset_partition_mode),
        migrations.CreateModel(
            name="OCPAllCostLineItemProjectDailySummaryP",
            fields=[
                ("id", models.UUIDField(serialize=False)),
                ("source_type", models.TextField()),
                ...
            ]
        ),
        migrations.RunPython(code=unset_partition_mode, reverse_code=set_partition_mode),
    ]

It's a little complicated, but it does work.

I hope that this can be adapted to your projects and open up the possibility of partitioning to those projects that need it.

Efficient Django Delete Cascade

HAP — Fri, 02 Jul 2021 12:11:59 +0000

Background

I'm working on a project that is developed using django. Part of the is project requires deleting n-level-deep relational data with a huge quantity of detail records at the lowest level.

Django allows for the setting of on-delete actions in the ForeignKey instantiation. Unfortunately, the on-delete action is not set in the database. It turns out that django is designed to handle cascade actions in the ORM at the application level. What this means is that it can use up a lot of memory resources as it walks relations, instantiates objects and deletes them through its ORM functionality.

Our project is deployed in Openshift. This ORM activity was causing the pod to be halted as it OOM'd. This was a drag for everyone and was causing data processing operations to back up.

So, what to do about it? I decided to not worry about django messaging at the lower cascade levels and instead manage the cascade myself. This required the development of some utility functions to compile QuerySet instances into SQL and another function to recursively walk model relations. The idea was to have the relation walker recursively reach the lowest relations level, then execute SQL directly on its way back up. Also, there is no query execution until the recursion works its way back. The benefit here is that we can use the ORM to construct a top-level query for records to delete, then the cascade function will find all of the models that depend on records from the target and execute the SQL necessary to simulate the delete cascade functionality that a database like PostgreSQL can handle.

This design was intended to be a compromise for ease of use in our application and not have to write a complicated django application extension or change the entire applications model inheritance to use a subclass of models.Model.

Design

Utility Functions

This required the creation of small set of utility functions to handle compilation of a QuerySet instance into either DELETE or UPDATE SQL statements. Also, a common function was needed to directly execute the compiled SQL with parameters

The delete SQL compiler:

from django.db import transaction
from django.db.models.sql.compiler import SQLDeleteCompiler


def get_delete_sql(query):
    """
    Compile a DELTE SQL statement from a QuerySet instance
    Params:
        query (QuerySet) : The query to compile to SQL
    Returns (tuple):
        (sql, params) : sql is the sql statement to exeucte
                        params are any parameters for the statement to use. This is a tuple.
    """
    return SQLDeleteCompiler(query.query, transaction.get_connection(), query.db).as_sql()

The update SQL compiler:

from django.db import models


def get_update_sql(query, **updatespec):
    """
    Compile the query with the update specifications into an UPDATE SQL statement and parameters
    Params:
        query (QuerySet) : The QuerySet that will select the row(s) to update
        updatespec (dict) : {column: new_value} expressed in the function call as `column=new_value` named parameters
    Returns (tuple):
        (sql, params) : sql is the sql statement to exeucte
                        params are any parameters for the statement to use. This is a tuple.
    """
    assert query.query.can_filter()
    query.for_write = True
    q = query.query.chain(models.sql.UpdateQuery)
    q.add_update_values(updatespec)
    q._annotations = None

    return q.get_compiler(query.db).as_sql()

The executor:

import logging

from django.db import transaction
from sqlparse import format as format_sql


LOG = logging.getLogger(__name__)


def execute_compiled_sql(sql, params=None):
    """
    Execute the SQL with any parameters directly using connection.cursor()
    Params:
        sql (str) : The parameterized SQL statement.
        params (tuple/list) : Parameters for the sql statement or None
    Returns :
        int : rows affected by the statement
    """
    rows_affected = 0
    with transaction.get_connection().cursor() as cur:
        params = params or None
        LOG.debug(format_sql(cur.mogrify(sql, params).decode("utf-8"), reindent_aligned=True))
        cur.execute(sql, params)
        rows_affected = cur.rowcount

    return rows_affected

This executor function will log the actual sql statement (with parameters inline) at the DEBUG level.

Cascade Function

The cascade function will walk all relations defined in the django model and will directly execute CASCADE and SET_NULL on-delete actions. Anything else is passed to the database and will return any database exception needed on error.

To call this function, all you need is the model from which you wish to start the cascade actions and a query against that model that will return the row(s) you wish to delete.

def cascade_delete(from_model, instance_pk_query, skip_relations=[], base_model=None, level=0):
    """
    Performs a cascading delete by walking the Django model relations and executing compiled SQL
    to perform the on_delete actions instead or running the collector.
    Parameters:
        from_model (models.Model) : A model class that is the relation root
        instance_pk_query (QuerySet) : A query for the records to delete and cascade from
        base_model (None; Model) : The root model class, If null, this will be set for you.
        level (int) : Recursion depth. This is used in logging only. Do not set.
        skip_relations (Iterable of Models) : Relations to skip over in case they are handled explicitly elsewhere
    """
    if base_model is None:
        base_model = from_model
    instance_pk_query = instance_pk_query.values_list("pk").order_by()
    LOG.info(f"Level {level} Delete Cascade for {base_model.__name__}: Checking relations for {from_model.__name__}")
    for model_relation in from_model._meta.related_objects:
        related_model = model_relation.related_model
        if related_model in skip_relations:
            LOG.info(f"SKIPPING RELATION {related_model.__name__} from caller directive")
            continue

        if model_relation.on_delete.__name__ == "SET_NULL":
            filterspec = {f"{model_relation.remote_field.column}__in": models.Subquery(instance_pk_query)}
            updatespec = {f"{model_relation.remote_field.column}": None}
            LOG.info(
                f"    Executing SET NULL constraint action on {related_model.__name__}"
                f" relation of {from_model.__name__}"
            )
            rec_count = execute_update_sql(related_model.objects.filter(**filterspec), **updatespec)
            LOG.info(f"    Updated {rec_count} records in {related_model.__name__}")
        elif model_relation.on_delete.__name__ == "CASCADE":
            filterspec = {f"{model_relation.remote_field.column}__in": models.Subquery(instance_pk_query)}
            related_pk_values = related_model.objects.filter(**filterspec).values_list(related_model._meta.pk.name)
            LOG.info(f"    Cascading delete to relations of {related_model.__name__}")
            cascade_delete(
                related_model, related_pk_values, base_model=base_model, level=level + 1, skip_relations=skip_relations
            )

    LOG.info(f"Level {level}: delete records from {from_model.__name__}")
    if level == 0:
        del_query = instance_pk_query
    else:
        filterspec = {f"{from_model._meta.pk.name}__in": models.Subquery(instance_pk_query)}
        del_query = from_model.objects.filter(**filterspec)

    rec_count = execute_delete_sql(del_query)
    LOG.info(f"Deleted {rec_count} records from {from_model.__name__}")

Note the skip_relations override. This is necessary for special cases that are handled outside of the cascade process and would either be unnecessary work or would possibly introduce an error. For my project, it was to skip partitioned tables where the delete involved simply dropping a partition. This would be unnecessary work and would be inefficient since all partitions could be scanned.

So the cascade will be handled appropriately as it checks all relations at every level and executes deletes bottom-up. This allows us to use the existing model relations and allows us to use the ORM QuerySet to build queries as normal.

I hope that you will find this useful if you are also using django and need to delete large amounts of related data via cascading actions.

Addendum

The project I'm working on makes heavy use of deferred constraints. If you are using this cascade delete and your database has deferred constraints, you may want to set them to immediate for this transaction.