django - Schinckel.net

Collecting Javascript Coverage in Django

2022-10-26 @ 20:44:33

here.

Over the past years, I’ve worked really hard to get code coverage of my work project up to a reasonable level. I’m not attempting to get 100% coverage (although, I have made it there in some smaller projects), nor do I mistakenly believe that 100% coverage is a panacea. I’m aware that even 100% coverage doesn’t mean there are no bugs.

But there are a lot of reasons to have high test coverage. Having less than 100% coverage mean that the code not being covered is not automatically tested.

In a django project, there are several different types of code. Python code, HTML, CSS and Javascript. In my case, there’s also a lot of SQL. Some of it is doing things like defining views and functions - but I’m not going to deal with that now…

What I am going to look at is Javascript.

It would be awesome to write unit tests for all of our JS code, but we’ve also been writing Robot Framework tests for a lot of our functionality: what would be great is to capture the coverage that is already happening due to this. It would be even better if it was possible to get this coverage information for JS that is embedded in HTML templates, but that seems like it might be a harder job.

There are really only two things we need to do in order to capture the coverage data:

Modify the scripts so they generate coverage data when they are executed
Capture the coverage data before any navigation away from any page

The first one could be done in a number of different ways. You might be able to use a staticfiles backend to modify the files when running collectstatic, or perhaps a middleware that modified the files as they are served. That could possibly modify the inline script tags as well, although ideally you’d want to be doing that in the template rendering phase, as I think that could be the last point you get access to the template file names and locations.

There is a tool that you can use for the actual modification of the script files: nyc. This used to be called Istanbul, I believe, and it can “instrument” files. Then, when the script is executed by the browser, it stores the coverage inforamtion in window.__coverage__.

This then allows for capture of the coverage information, but how do we ensure this information is not lost when the page is navigated away from?

To do that in Django requires a small amount of JS, and a django view to capture it:

window.addEventListener('beforeunload', () => {
  if (window.__coverage__) {
    navigator.sendBeacon("{ % url 'collect-coverage' % }", JSON.stringify(window.__coverage__));
  }
})

Okay, that was easier than expected…what about the matching view?

import json
import uuid
from pathlib import Path

from django.http import HttpResponse

@csrf_exempt
def collect_coverage(request):
    coverage_dir = Path('js-coverage')
    if not coverage_dir.exists():
        coverage_dir.mkdir()

    try:
        json.dump(
            json.loads(request.data),
            (coverage_dir / 'coverage.{uuid.uuid4()}.json').open('w')
        )
    except ValueError:
        pass

    return HttpResponse('')

Oh, that wasn’t too bad either. It just checks that the incoming data is valid JSON, and then dumps it out to a uniquely named JSON file in js-coverage/.

We’ll need to install that in our urls.py:

if settings.DEBUG or settings.TESTING:
    path('coverage/', collect_coverage, name='collect-coverage')

Note well that you will need to ensure you don’t expose this view when not running your Selenium/Robot tests, else someone could very quickly fill up your disk with JSON files…

The last bit of the puzzle is to merge the various files together to generate the report:

nyc report --reporter lcov -t .js-coverage --report-dir .

This generates an lcov.info file that could be uploaded to a coverage analysis tool, and an lcov-report/ directory that contains an HTML report.

So, what about the tricky part, the bit that actually handles running the nyc instrument command on the files?

There are two types of scripts we need to be able to mutate: standalone files (ie, collectstatic), and inline scripts (within HTML templates).

Static files storage backends are relatively well-known, here’s some code that might work:

class InstrumentJSBackend(django.contrib.staticfiles.storage.StaticFilesStorage):
    def post_process(self, paths, dry_run=False, **options):
        if dry_run:
            return

        for path in paths:
            if path.endswith('.js'):
                # We need the original path name, not the one we have just copied...

            yield path, path, True

The latter can be done by using a custom template loader; it’s quite similar in construct to the supplied caching loader:

class Loader(BaseLoader):
    def __init__(self, engine, loaders):
        self.loaders = engine.get_template_loaders()
        super().__init__(engine)

    def get_dirs(self):
      for loader in self.loaders:
          if hasattr(loader, 'get_dirs'):
              yield from loader.get_dirs()

    def get_contents(self, origin):
        # Rewrite any script tags using `nyc instrument`
        contents = origin.loader.get_contents(origin)
        return self.rewrite(origin, contents)

    def get_template_sources(self, template_name):
        for loader in self.loaders:
            yield from loader.get_template_sources(template_name)

    def rewrite(self, origin, contents):

        return subprocess.check_output(['nyc', 'instrument', origin])

Django Implied Relationship

2021-07-14 @ 17:35:43

here.

[This post was updated on 2021-07-28].

A little while ago, Alec McGavin put up a post on the Kogan blog about Custom Relationships in Django. This is a really cool way to get a relationship in Django that exists in the database, but cannot be modelled correctly in Django. For instance, this could be data in the database that does not have a Foreign Key, either because it’s legacy data, or because it’s possible either side of the relationship might be added to the database before the other, rather than the normal order in a database where the target table of the FK is always added to first.

However, I have another slighly different situation where an implied relationship exists, but should not be stored directly.

Consider the following data structures:

class Employee(models.Model):
    name = models.TextField()


class EmploymentPeriod(models.Model):
    employee = models.ForeignKey(
        Employee,
        related_name='employment_periods',
        on_delete=models.CASCADE,
    )
    valid_period = DateRangeField()

    class Meta:
        constraints = [
            ExclusionConstraint(
                name='employment_overlap',
                expressions=[
                    ('employee', RangeOperators.EQUAL),
                    ('valid_period', RangeOperators.OVERLAPS),
                ]
            )
        ]


class Shift(models.Model):
    employee = models.ForeignKey(
        Employee,
        related_name='shifts',
        on_delete=models.CASCADE,
    )
    date = models.DateField()
    start_time = models.TimeField()
    duration = models.DurationField()
    employment_period = Relationship(
        EmploymentPeriod,
        from_fields=['employee', 'date'],
        to_fields=['employee', 'valid_period'],
    )

    @property
    def start(self):
        return datetime.datetime.combine(self.date, self.start_time)

    @property
    def finish(self):
        return self.start + self.duration

Now, there is a direct relationship between Shift and Employee, and also between EmploymentPeriod and Employee, but there could be an inferred relatonship between Shift and EmploymentPeriod. Because of the exclusion constraint, we know there will only be one EmploymentPeriod for a given Shift.

It would be really nice to be able to create this relationship, so we can reference the employment period (or lack thereof) directly. The Relationship class above goes close, but tries to use an equality check between the date and date_range fields.

It turns out, we can add a bit to that class, and teach it how to handle the various ways this relationship can be accessed:

Shift().employment_period
EmploymentPeriod().shifts
Shift.objects.select_related(’employment_period')
EmploymentPeriod.objects.prefetch_related(‘shifts’)
Shift.objects.filter(employment_period=emp)
Shift.objects.filter(employment_period__in_=[emp1, emp2])
Shift.objects.filter(employment_period__isnull=True)
Shift.objects.filter(employment_period=None)
Shift.objects.filter(employment_period__isnull=False)
EmploymentPeriod.objects.filter(shifts__contains=shift)
EmploymentPeriod.objects.filter(shifts__contains=[shift1, shift2])
EmploymentPeriod.objects.filter(shifts__isnull=True)
EmploymentPeriod.objects.filter(shifts__isnull=False)

…plus there is also the inverse of a bunch of these - ie Shift.objects.exclude(employment_period=emp). In some cases these are equivalent, but that’s not always possible to determine.

Let’s have a look at the original class, and a new subclass for these non-direct relationships:

from django.db import models
from django.db.models.lookups import Lookup


class Relationship(models.ForeignObject):
    """
    Create a django link between models on a field where a foreign key isn't used.
    This class allows that link to be realised through a proper relationship,
    allowing prefetches and select_related.

    https://devblog.kogan.com/blog/custom-relationships-in-django
    """

    def __init__(self, model, from_fields, to_fields, **kwargs):
        super().__init__(
            model,
            on_delete=models.DO_NOTHING,
            from_fields=from_fields,
            to_fields=to_fields,
            null=True,
            blank=True,
            **kwargs,
        )

    def contribute_to_class(self, cls, name, private_only=False, **kwargs):
        # override the default to always make it private
        # this ensures that no additional columns are created
        super().contribute_to_class(cls, name, private_only=True, **kwargs)

There’s not much to this, which is part of the beauty of it. Django pretty-much handles composite-key relationships, it just won’t create actual ForeignKeys based on them. There have been noises about implementing this for years, and maybe eventually it will happen.

But what about a subclass that allows the implicit relationship described above?

class ImplicitRelationship(Relationship):
    """
    Create a django link between two models where at least one of the fields
    uses a containment (or other type of non-direct) relationship.

    Relationship should be used if this is just a composite key (or a single
    key that is not a real ForeignKey in the database).

    """

    def get_path_info(self, filtered_relation=None):
        """Get path from this field to the related model."""
        opts = self.remote_field.model._meta
        from_opts = self.model._meta
        self.related_fields
        return [
            PathInfo(
                from_opts=from_opts,
                to_opts=opts,
                target_fields=[rhs for lhs, rhs in self.other_related_fields],
                join_field=self,
                m2m=False,
                direct=False,
                filtered_relation=filtered_relation,
            )
        ]

    def get_reverse_path_info(self, filtered_relation=None):
        """Get path from the related model to this field's model."""
        opts = self.model._meta
        from_opts = self.remote_field.model._meta
        self.related_fields
        return [
            PathInfo(
                from_opts=from_opts,
                to_opts=opts,
                target_fields=[lhs for lhs, rhs in self.other_related_fields],
                join_field=self.remote_field,
                m2m=False,
                direct=False,
                filtered_relation=filtered_relation,
            )
        ]

    @cached_property
    def other_related_fields(self):
        return self.resolve_related_fields()

    @cached_property
    def related_fields(self):
        return []

    def get_local_related_value(self, instance):
        """
        Given an instance, determine the value that will be used as
        the key for this value in a dict of related items.

        This is where it starts to get tricky. Django only really expects
        keys to match exactly, but we may have a value that contains a
        date, that needs to be checked for inclusion in a DateRange.

        Whilst psycopg2 does not normalise Range values, it will handle
        <date> in <DateRange> correctly, so we can use that as the
        comparison.
        """
        parts = self.get_instance_value_for_fields(
            instance,
            [lhs for lhs, rhs in self.other_related_fields],
        )

        if not hasattr(self, '_known_instance_keys'):
            return parts

        if parts in self._known_instance_keys:
            return parts

        for keys in self._known_instance_keys:
            for part, key in zip(parts, keys):
                if part == key or getattr(key, '__contains__', None) and part in key:
                    return keys

    @property
    def get_foreign_related_value(self):
        """
        Because we need to use non-exact matching, we need to set up to store
        instances based on known keys. The Django code that uses this builds up
        a dict of keys/values, but since we need to do containment testing in
        get_local_related_value(instance), we have to store a local set of
        key values, which will be used for that containment checking.

        This is implemented as a property that returns a function, after clearing
        out the cache of known instances so that each queryset will have it's own
        cache. Otherwise, instances from the last run through would be matched
        in the next run.

        This uses knowledge of the Django internals, where this method is called
        before get_local_related_value, which really is not ideal, but there does
        not seem to be a better way to handle this.
        """
        self._known_instance_keys = set()

        def get_foreign_related_value(instance):
            values = self.get_instance_value_for_fields(
                instance, 
                [rhs for lhs, rhs in self.other_related_fields]
            )

            self._known_instance_keys.add(values)
            return values

        return get_foreign_related_value

    def get_extra_restriction(self, where_class, alias, remote_alias):
        """
        This method is used to get extra JOIN conditions.

        We don't need to include the exact conditions, only those
        that we filtered out from the regular related_fields.
        The exact conditinos will be already applied to the JOIN
        by the regular handling.
        """
        if not alias or not remote_alias:
            return

        if self.other_related_fields:
            cond = where_class()

            for local, remote in self.other_related_fields:
                local, remote = local.get_col(remote_alias), remote.get_col(alias)
                lookup_name = JOIN_LOOKUPS.get(get_key(local, remote), 'exact')
                lookup = local.get_lookup(lookup_name)(local, remote)
                cond.add(lookup, 'AND')

            return cond

    def get_extra_descriptor_filter(self, instance):
        """
        The sibling to get_extra_restriction, this one is used to get the extra
        filters that are required to limit to the correct objects.
        """
        extra_filter = {}
        for lhs, rhs in self.other_related_fields:
            lookup = JOIN_LOOKUPS.get(get_key(rhs.cached_col, lhs.cached_col), 'exact')
            extra_filter[f'{rhs.name}__{lookup}'] = getattr(instance, lhs.attname)
        return extra_filter

    def get_where(self, value, alias=None):
        constraint = WhereNode(connector=AND)

        values = self.get_instance_value_for_fields(
            value, 
            [remote for local, remote in self.other_related_fields]
        )

        for (source, target), value in zip(self.other_related_fields, values):
            key = (source.get_internal_type(), target.get_internal_type())
            lookup_type = JOIN_LOOKUPS.get(key, 'exact')
            lookup_class = source.get_lookup(lookup_type)
            lookup = lookup_class(target.get_col(alias or self.model._meta.db_table, source), value)
            constraint.add(lookup, AND)

        return constraint

    def get_exists_subquery_filters(self, inverted=False):
        filters = {}
        for source, target in self.other_related_fields:
            if inverted:
                source, target = target, source
            key = (target.get_internal_type(), source.get_internal_type())
            lookup = JOIN_LOOKUPS.get(key, 'exact')
            filters[f'{target.attname}__{lookup}'] = OuterRef(source.attname)
        return filters

    def exists_subquery(self, negated=False, inverted=False, **filters):
        if inverted:
            exists = Exists(
                self.model.objects.filter(
                    **self.get_exists_subquery_filters(inverted=True), 
                    **filters
                ).values('pk')
            )
        else:
            exists = Exists(
                self.related_model.objects.filter(
                    **self.get_exists_subquery_filters(), 
                    **filters
                ).values('pk')
            )
        return ~exists if negated else exists

    @property
    def target_field(self):
        raise FieldError()


# We should be able to add more pairs here as we need to handle them.
JOIN_LOOKUPS = {
    ('DateField', 'DateRangeField'): 'contained_by',
    ('DateRangeField', 'DateField'): 'contains',
}

def get_key(lhs, rhs):
    return (lhs.output_field.get_internal_type(), rhs.output_field.get_internal_type())

There’s actually a lot more code there than I really wanted, however it seems mostly to be necessary.

But wait, there’s more. We also need to teach Django how to handle the various lookups that can be performed on these relationships:

@ImplicitRelationship.register_lookup
class RelatedMultipleExact(Lookup):
    lookup_name = 'exact'
    """
    Apply each lookup type from each of the fields in an ImplicitRelationship.

    This is for querysets of the form:

    >>> Foo.objects.filter(relationship=instance)

    This gets the relevant operator for each of the lookups, based on
    the field type of the pair of (to/from) fields.
    """

    def as_sql(self, compiler, connection):
        field = self.lhs.field

        # If we only have a primary key here, and not an instance, then we
        # will need to push the querying back into the database - normally
        # a diroct lookup just uses the value as a PK, but here we need to
        # get the database to do a subquery to get the other values.
        if self.rhs_is_direct_value() and not isinstance(self.rhs, models.Model):
            # We can't unref the alias here, because Django will have also put in a IS NOT NULL
            # on the thing, which is referencing the wrong table.
            compiler.query.alias_map[self.lhs.alias] = compiler.query.alias_map[self.lhs.alias].promote()
            return field.exists_subquery(pk=self.rhs).resolve_expression(compiler.query).as_sql(compiler, connection)

        return field.get_where(self.rhs, alias=self.lhs.alias).as_sql(compiler, connection)


@ImplicitRelationship.register_lookup
class RelatedMultipleIn(Lookup):
    """
    Apply each lookup from each of the fields in an ImplicitRelationship.

    This is for querysets of the form:

    >>> Foo.objects.filter(relationship__in=[instance1, instance2])
    >>> Foo.objects.filter(relationship__not_in=[instance1, instance2])

    This builds an EXISTS() clause that uses a subquery to find
    if each instance matches - this is usually better than a bunch
    of clauses that would use OR, because that would preclude the
    use of indexes.
    """

    lookup_name = 'in'
    negated = False

    def as_sql(self, compiler, connection):
        field = self.lhs.field
        if self.negated:
            # We remove one reference to the joined table, so that if we only
            # have this reference, ie no columns, then we don't even join the
            # table in (as we'll be using an EXISTS in WHERE)
            compiler.query.unref_alias(self.lhs.alias)
        return (
            field.exists_subquery(
                negated=self.negated,
                pk__in=[getattr(x, 'pk', x) for x in self.rhs],
            )
            .resolve_expression(compiler.query)
            .as_sql(compiler, connection)
        )


@ImplicitRelationship.register_lookup
class RelatedMultipleNotIn(RelatedMultipleIn):
    negated = True
    lookup_name = 'not_in'


@ImplicitRelationship.register_lookup
class RelatedMultipleNull(Lookup):
    """
    Apply each lookup from each of the fields in an ImplicitRelationship.

    This is for querysets of the form:

    >>> Foo.objects.filter(relationship=None)
    >>> Foo.objects.exclude(relationship=None)
    >>> Foo.objects.filter(relationship__isnull=True)
    >>> Foo.objects.filter(relationship__isnull=False)

    """

    lookup_name = 'isnull'

    def as_sql(self, compiler, connection):
        field = self.lhs.field

        if not isinstance(field, Relationship):
            pk = field.related_model._meta.pk
            lookup = pk.get_lookup('isnull')(pk.get_col(self.lhs.alias), self.rhs)
            return lookup.as_sql(compiler, connection)

        # We remove one reference to the joined table, so that if we only
        # have this reference, ie no columns, then we don't even join the
        # table in (as we'll be using an EXISTS in WHERE)
        compiler.query.unref_alias(self.lhs.alias)
        return field.exists_subquery(negated=self.rhs).resolve_expression(compiler.query).as_sql(compiler, connection)


@ImplicitRelationship.register_lookup
class RelatedMultipleContains(Lookup):
    """
    Apply each lookup from each of the fields in an ImplicitRelationship.

    This is for querysets of the form:

    >>> Foo.objects.filter(reverse_relationship__contains=[x, y])
    >>> Foo.objects.filter(reverse_relationship__contains=x)

    """

    lookup_name = 'contains'
    negated = False

    def as_sql(self, compiler, connection):
        if isinstance(self.lhs.field, Relationship):
            raise TypeError(f'Unable to perform __{self.lookup_name} queries on Relationship, only on reversed')

        field = self.lhs.field.remote_field

        try:
            iter(self.rhs)
        except TypeError:
            value = getattr(self.rhs, 'pk', self.rhs)
            lookup = 'pk'
        else:
            if self.negated:
                raise ValueError('Unable to perform not_contains=list')
            value = [getattr(x, 'pk', x) for x in self.rhs]
            lookup = 'pk__in'

        # We remove one reference to the joined table, so that if we only
        # have this reference, ie no columns, then we don't even join the
        # table in (as we'll be using an EXISTS in WHERE)
        compiler.query.unref_alias(self.lhs.alias)

        return (
            field.exists_subquery(negated=self.negated, inverted=True, **{lookup: value})
            .resolve_expression(compiler.query)
            .as_sql(compiler, connection)
        )


@ImplicitRelationship.register_lookup
class RelatedMultipleNotContains(RelatedMultipleContains):
    lookup_name = 'not_contains'
    negated = True

There is one caveat to this - these lookups (under certain conditions) will change the list of FROM tables that are required to be joined into the query. In some cases this adds a required JOIN, in others it actually removes the need for a JOIN because the WHERE clause is all within a subquery, and should not contain a join to the original table.

However, until [https://github.com/django/django/pull/14683](PR 14683) is merged, these lookups will not always work - if the list of tables is mutated by the lookup, the SQL query that is generated will not contain this required mutation. Hopefully I can get that PR merged, but it is possible to use patchy to patch your local installation until that is done.

Extending Django Wizards

2021-04-23 @ 20:12:51

here.

I quite like the wizard pattern. Most people are somewhat familiar with the concept of having to complete a logically grouped set of form elements before moving onto the next one, with a final confirmation step, after which the operation is committed.

Howver, there are some perceieved problems within the Django formtools implementation, at least from the perspective of our use case.

The big one my team have identified: when you reload the page with a GET request, it clears out the data from the current run of the wizard. This is problematic when something happens to your internet, and you have to reload a page manually, for instance. Or if you just happen to reload it through another means.

Related to this, if you don’t have all of the information you require, there’s no way to “stash” your current iteration, and return to it later. This later could be after a logout and login (or even on a different computer).

Additionally, there’s no way to have “extra” steps inserted based on the selections made in a previous step. You could take the other approach, and define all of the steps initially (and then skip those that aren’t required), but in our case, we need a repeatable step (I call them a sub-step), where we perform an operation on a bunch of items that were selected in a previous step.

Let’s look at each of these in turn.

The default implementation automatically clears the storage whenever a GET request is handled:

class WizardView(TemplateView):

    def get(self, request, *args, **kwargs):
        self.storage.reset()
        self.storage.current_step = self.steps.first
        return self.render(self.get_form())

We can replace this with an implementation that will only reset when it detects a special GET parameter:

    def get(self, request, *args, **kwargs):
        if not request.GET.get('reset'):
            self.storage.reset()
            self.storage.current_step = self.steps.first
        
        step = self.storage.current_step or self.steps.first
        return self.render(self.get_form(
            step,
            data=self.storage.get_step_data(step),
            files=self.storage.get_step_files(step),
        ))

Not that we need to build up the form data to provide to the view - otherwise when you load up the wizard, it will render the first page as empty, but if you then use the navigation to select the first page it will correctly render it with data.

You’ll need to have an explicit link or button in your wizard template(s) to enable the user to restart the wizard if they need to.

Next up is persistent storage. For this we will need somewhere to store the data. The formtools implementation makes this easy to swap out - you can define the storage backend you want to use.

class MyWizard(WizardView):
    form_list = [...]
    storage_name = 'storage.wizard.DatabaseStorage'

That’s all you need to do to make your wizard use it - but now we need to build the storage class.

Let’s begin with a model to store the data in.

class WizardStorage(models.Model):
    user = models.ForeignKey('auth.User', related_name='wizard_storage', on_delete=models.CASCADE)
    prefix = models.TextField()
    data = models.JSONField()

We could use extra fields for the various parts, but that just complicates things.

Now let’s see a storage implementation:

class DatabaseStorage(BaseStorage):
    def __init__(self, prefix, request=None, file_storage=None):
        super().__init__(prefix, request, file_storage)
        self.init_data()
    
    def init_data(self):
        self.instance, _create = WizardStorage.objects.get_or_create(
            user=self.request.user,
            prefix=self.prefix,
            defaults={
                'data': {
                    self.step_key: None,
                    self.step_data_key: {},
                    self.step_files_key: {},
                    self.extra_data_key: {},
                }
            }
        )
    
    @property
    def data(self):
        return self.instance.data
    
    def update_response(self, response):
        if hasattr(self, 'instance'):
            self.instance.save()
      
    def reset(self):
        if hasattr(self, 'instance'):
            self.instance.delete()
            del self.instance

You can make it a bit more configurable so that the model could be swapped quite easily.

From here, we can use this in a wizard, and it will persist the step data to the database. I’ve still got a bit to do to ensure it can handle files, but this has not been a requirement of mine as yet.

There could be a bit of fun around having multiple stashes for a given wizard, and allowing the user to select which one they want to work on. As it stands, it just uses the user id and wizard prefix to determine where the data is stored.

The third improvement, allowing sub-steps, is a bit more complicated. To do that, you need to replace a bit more of the internal formtools code, rather than just subclassing/extending it. This involves a bunch of patching of the StepsHelper class from formtools - although you could replace this class by overriding WizardView.dispatch.

That’s beyond the scope of this post.

Django modify_settings and receivers

2021-03-31 @ 18:47:08

here.

Sometimes, tests expose weird behaviour.

In this instance, I have a Makefile command that calls a cookiecutter command to create a new integration with a payroll system. We’d noticed that over time, this infrequntly used code had become slightly less than perfect.

So, I wrote a test that runs this code. Not directly using the make new-payroll-system command, but rather using a test case from within Django.

This worked great in development. I was able to have code that was generated, some tests on that code were run (including generating migrations for the new app), and then the code was removed.

def test_cookie_cutter(self):
    from cookiecutter.main import cookiecutter
    
    try:
        cookiecutter(
            'integrations/__template__', 
            extra_context={'system_name': 'new_thing'}, 
            output_dir='integrations/systems',
            no_input=True,
        )
        with modify_settings(INSTALLED_APPS={'append': 'integrations.systems.new_thing'}):
            call_command('makemigrations', 'new_thing', no_input=True, verbosity=0)
    finally:
        shutil.rmtree('integrations/systems/new_thing')

But this failed in CI.

Turns out that our Codeship-based testing infrastructure doesn’t allow for writing the files in the expected location.

Never mind, we can use tempfile.TemporaryDirectory() instead. That will handle the cleanup for us, which is better than removing files ourselves:

def test_cookie_cutter(self):
    from cookiecutter.main import cookiecutter
    
    with tempfile.TemporaryDirectory() as dirname:
        # Put our new temporary directory on the PYTHONPATH.
        sys.path.insert(0, dirname)
        
        cookiecutter(
            'integrations/__template__', 
            extra_context={'system_name': 'new_thing'}, 
            output_dir=dirname,
            no_input=True,
        )
        with modify_settings(INSTALLED_APPS={'append': 'integrations.systems.new_thing'}):
            call_command('makemigrations', 'new_thing', no_input=True, verbosity=0)

There’s a little more to my code, but it’s not really relevant.

What is revelant is that, whilst this test was working fine, there is a subsequent test that was failing. Because, as part of the cookie-cutter template, we install a signal handler by default, and this signal handler was still connected even after the app was removed by the end of the modify_settings context manager.

So, how can we remove the signal handler when we are done? It should be possible to, in the context manager, look at which signals exist before we run, and then compare that to the ones that are connected when we are exiting…

Turns out, we don’t need to. In this context, because we aren’t really running the code from the new django app, we can just prevent the signal handlers being connected in the first place:

@patch('django.dispatch.receiver')
def test_cookie_cutter(self, _receiver):
    from cookiecutter.main import cookiecutter
    
    with tempfile.TemporaryDirectory() as dirname:
        # Put our new temporary directory on the PYTHONPATH.
        sys.path.insert(0, dirname)
        
        cookiecutter(
            'integrations/__template__', 
            extra_context={'system_name': 'new_thing'}, 
            output_dir=dirname,
            no_input=True,
        )
        with modify_settings(INSTALLED_APPS={'append': 'integrations.systems.new_thing'}):
            call_command('makemigrations', 'new_thing', no_input=True, verbosity=0)

That one change of mocking out the @receiver decorator means that it won’t attach the signals that it comes across when doing the modify_settings, which is exactly what I want to happen in this case.

Cleaning up a required process when you quit

2021-01-27 @ 14:32:53

here.

Some time ago, I moved a bunch of our devops tools into a Makefile. I’m not sure exactly what the benefit was then, but there have been some challenges.

One of these was related to having to run multiple processes when performing a task, but cleaning up the dependencies when we finish our task. Specifically, we run a websocket server in production that uses Redis for PubSub. We can safely assume the developer is running redis already, since the rest of the platform also depends upon it. However, to collect the changes from Postgres and publish them into Redis, we use a custom django management command. All this does is listen for postgres notifications and forward them to redis.

$ ./manage.py relay

Mostly we don’t keep this running in development, unless you need to be running the websocket code locally for testing. It is a hassle to remember to start a process and then kill it when you are done, so it would be convenient to start the relay command, then start our websocket worker, and clean up both processes when we quit the worker.

This can be done in a Makefile command:

websocket:
    $(shell trap 'kill 0' SIGINT ; \
        ./manage.py relay & \
        gunicon \
          -k geventwebsocket.gunicorn.workers.GeventWebsocketWorker \
          -workers=1 \
          -reload \
          webocket:application \
    )

The trap command performs the magic - it will kill the background command when the gunicorn worker is killed with Ctrl-C.

I have some other commands which start the background process, and then use kill -0 $(shell ps ax | grep ... | head -n 1 | cut -d ' ' -f 1) to ensure they kill the process on quit, but this sometimes does not work: I should get around to changing them over to this…

Scheduling Celery Tasks in the (far) future

2021-01-19 @ 19:59:47

here.

We used to make use of the fact that a celery task can be scheduled at some time in the future to auto-punch-out staff members who failed to punch out 24 hours after their shift started. This was as simple as scheduling a task with an eta=86400.

However, as Adam points out here (see number 5). This is not a great idea. For one, it will bog down your celery workers (and make them use a bunch more memory). Secondly, something Adam doesn’t mention is that if your queue is corrupted, then all of those future tasks will not be executed.

Discussing this in IRC today, I thought of a simple mechanism for scheduling these tasks and processing them at some point after they are due.

We will start with a model to store our task:

class ScheduledTask(models.Model):
    task_name = models.TextField()
    task_args = models.JSONField(default=list)
    task_kwargs = models.JSONField(default=dict)
    due_date = models.DateTimeField()

    objects = ScheduledTaskQuerySet.as_manager()

    @property
    def task(self):
        module, task = task_name.rsplit('.', 1)
        return getattr(importlib.import_module(module), task)

    def execute(self):
        self.task.apply_async(args=self.args, kwargs=self.kwargs)

We have a custom queryset defined, that allows us to see which tasks are due using a nice queryset method:

class ScheduledTaskQuerySet(models.query.QuerySet):
    def due(self):
        return self.due_date < datetime.datetime.utcnow()

    def schedule(self, task, when, *args, **kwargs):
        self.create(task_name=task_name, task_args=args, task_kwargs=kwargs, due_date=when)

Finally, we need a task that will enqueue the due tasks:

@app.task
def enqueue_due_tasks():
    for task in ScheduledTask.objects.due():
        task.execute()
        task.delete()

As it stands, with this code, to schedule a task, you need to create a model instance:

ScheduledTask.objects.schedule(
    'foo.tasks.bar',
    datetime.datetime(2525, 11, 11, 9, 30),
    'x',
    y='z'
)

But, it would be nice if we could use the task to schedule itself:

foo.tasks.bar.schedule(args=[], kwargs={}, eta=X, countdown=Y)

Or, even better:

foo.tasks.bar.s(arg, kwarg=value).schedule(datetime.timedelta(days=365))

The first one we should be able to do by using custom tasks (and implementing a schedule method):

class Scheduled(celery.Task):
    def schedule(self, *, args=None, kwargs=None, eta=None, countdown=None):
        if not eta and not countdown:
            raise ValueError('One of eta and countdown must be supplied')
        if eta and countdown:
            raise ValueError('Only one of eta and countdown must be supplied')
        if eta:
            ScheduledTask.objects.schedule(self.name, eta, *args, **kwargs)
        else:
            ScheduledTask.objects.schedule(
                self.name,
                datetime.datetime.utcnow() + datetime.timedelta(countdown),
                *args, **kwargs
            )

Then, as long as a task is defined as using the base class, we can schedule it:

@app.task(base=Schedule)
def task_name(x, y=None):
    pass

But what about mucking with the .s() or .signature() calls? Now we are getting a bit experimental, but it still might be fun:

from celery.canvas import Signature

def schedule(self, when=None):
    if when:
        if isinstance(when, datetime.timedelta):
            when = datetime.datetime.utcnow() + when
    else:
        if self.options.countdown:
            when = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.options.countdown)
        elif self.options.eta:
            when = self.otions.eta

    ScheduledTask.objects.create(
        task_name=self.task,
        task_args=self.args,
        task_kwargs=self.kwargs,
        due_date=when,
    )

Signature.schedule = schedule

This is ~~totally~~ mostly untested, and I’m not sure it’s a really good idea.

A better idea might be to have a backend that doesn’t even tell the workers about the tasks until after they are due…that way you would not have to duck-punch celery at all.

RobotFramework, Chromedriver and Docker

2020-12-18 @ 23:42:26

here.

One of my team implemented RobotFramework support for automated browser testing of our platform a while ago. At the time, we were using Codeship Basic, and I built a helper to run a robot test suite within a tox environment. It was all good, because chromedriver and all it’s dependencies were already installed.

But time passes, and we needed to move to Codeship Pro. Which has some neater features, but required me to build docker images for everything. We already use docker for deployment, but I didn’t really want to build a bunch of distinct images just for testing that re-implemented the same stuff that we have in our deployment images. Even just appending new stuff to them means that things could turn out to be a pain in the arse to manage.

And getting chromedriver installed into a docker image is not neat.

I did find a docker image that just has an instance of chromedriver, and exposes that. But getting that to work with robot was still a bunch of work. After much experimentation, I was able to get the connections between everything to work.

First, we need to have the chromedriver container running:

$ docker run -p 4444:4444 CHROMEDRIVER_WHITELISTED_IPS='' robcherry/docker-chromedriver:latest

Then, there are a few moving parts that need to be in place to get things to work. Using my djangobot management command (which I had to extend a bit here), a single command can be used to spin up a Django runserver command, apply migrations (if necessary), and then run the robot commands. The trick is you need to teach Robot to speak to the remote WebDriver instance, which then in turn speaks to the running django webserver.

First, the RobotFramework commands; my resource.robot file which is referenced by all of my robot test suites contains:

*** Variables ***

${HOSTNAME}         127.0.0.1
${PORT}             8000
${SCHEME}           http
${SERVER}           ${SCHEME}://${HOSTNAME}:${PORT}
${BROWSER}          headlesschrome
${TIMEOUT}          30
${REMOTE_URL}

*** Settings ***

Documentation   A resource file with reusable keywords and variables.
Library         SeleniumLibrary             timeout=${TIMEOUT}      implicit_wait=1
Library         Collections
Library         DebugLibrary
Library         DateTime
Library         String
Library         djangobot.DjangoLibrary     ${HOSTNAME}     ${PORT}

*** Keywords ***

Create Remote Webdriver
    ${chrome_options} =     Evaluate    sys.modules['selenium.webdriver'].ChromeOptions()    sys, selenium.webdriver
    Call Method    ${chrome_options}   add_argument    headless
    Call Method    ${chrome_options}   add_argument    disable-gpu
    Call Method    ${chrome_options}   add_argument    no-sandbox
    ${options}=     Call Method     ${chrome_options}    to_capabilities

    Create Webdriver    Remote   command_executor=${REMOTE_URL}    desired_capabilities=${options}
    Open Browser    ${SERVER}   ${BROWSER}  remote_url=${REMOTE_URL}    desired_capabilities=${options}

Start Session
    Run Keyword If  '${REMOTE_URL}'    Create Remote Webdriver
    Run Keyword If  '${REMOTE_URL}' == ''    Open Browser    ${SERVER}   ${BROWSER}

    Set Window Size     2048  2048
    Fetch Url       login
    Add Cookie      robot   true

    Register Keyword To Run On Failure    djangobot.DjangoLibrary.Dump Error Data

End Session
    Close Browser

Logout
    Fetch Url     logout

Notice that the Start Session keyword determines which type of browser to open - either a local or remote one.

Thus, each *.robot file starts with:

*** Settings ***
Resource                resource.robot
Suite Setup             Start Session
Suite Teardown          End Session
Test Setup              Logout

Because the requests will no longer be coming from localhost, you need to ensure that your runserver is listening on the interface the requests will be coming from. If you can’t detect this, and your machine is not exposed to an insecure network, then you can use 0.0.0.0 to get the django devserver to listen on all interfaces. You will also need to supply the hostname that you will be using for the requests (which won’t be localhost anymore), and ensure this is in your Django settings.ALLOWED_HOSTS.

In my case, I needed to make my robot command allow all this, but ultimately I can now do:

$ ./manage.py robot --runserver 0 \
                    --listen 0.0.0.0 \
                    --hostname mymachine.local \
                    --remote-url http://localhost:4444 \
                    --include tag

This runs against the database I already have prepared, but in my codeship-steps.yml I needed to do a bit more, and hook it up to the other containers:

coverage run --branch --parallel \
    /app/manage.py robot --migrate \
                         --server-url=http://web.8000 \
                         --remote-url=http://chromedriver:4444 \
                         --tests-dir=/app/robot_tests/ --output-dir=/coverage/robot_results/ \
                         --exclude skip  --exclude expected-failure

Now, if only Codeship’s jet tool actually cached multi-stage builds correctly.

Maybe I neeed to try this.

Update value only if present

2020-12-18 @ 22:44:41

django
form

here.

We have a bunch of integrations with external systems, and in most of these cases we are unable to use Oauth, or other mechanisms that don’t require us to store a username password pair. So, we have to store that information (encrypted, because we need to use the value, rather than just being able to store a hashed value to compare an incoming value with).

Because this data is sensitive, we do not want to show this value to the user, but we do need to allow them to change it. As such, we end up with a form that usually contains a username and a password field, and sometimes a URL field:

class ConfigForm(forms.ModelForm):
    class Meta:
        model = ExternalSystem
        fields = ('username', 'password', 'url')

But this would show the password to the user. We don’t want to do that, but we do want to allow them to include a new password if it has changed.

In the past, I’ve done this on a per-form basis by overridding the clean_password method:

class ConfigForm(forms.ModelForm):
    class Meta:
        model = ExternalSystem
        fields = ('username', 'password', 'url')

    def clean_password(self):
        return self.cleaned_data.get('password') or self.instance.password

But this requires implementing that method on every form. As I mentioned before, we have a bunch of these. And on at least one, we’d missed this method. We could subclass a base form class that implements this method, but I think there is a nicer way.

It should be possible to have a field that handles this. The methods that look interesting are clean, and has_changed. Specifically, it would be great if we could just override has_changed:

class WriteOnlyField(forms.CharField):
    def has_changed(self, initial, data):
        return bool(data) and initial != data

However, it turns out this is not used until the form is re-rendered (or perhaps not at all by default, it’s very likely my code calls this to get a list of changed fields to mark as changed as a UI affordance).

The clean method in a CharField does not have access to the initial value, and there really is not a nice way to get this value attached to the field (other than doing it in the has_changed method, which is not called).

But it turns out this behaviour (apply changes only when a value is supplied) is the same behaviour that is used by FileField: and as such, it gets a special if statement in the form cleaning process, and is passed both the initial and the new values.

So, we can leverage this and get a field class that does what we want:

class WriteOnlyField(forms.CharField, forms.FileField):
    def clean(self, value, initial):
        return value or initial

    def has_changed(self, initial, data):
        return bool(data) and initial != data

We can even go a bit further, and rely on the behaviour of forms.PasswordInput() to hide the value on an unbound form:

class WriteOnlyField(forms.CharField, forms.FileField):
    def __init__(self, *args, **kwargs):
        defaults = {
            'widget': forms.PasswordInput(),
            'help_text': _('Leave blank if unchanged'),
        }
        defaults.update(**kwargs)
        return super().__init__(*args, **defaults)

    def clean(self, value, initial):
        return value or initial

    def has_changed(self, initial, data):
        return bool(data) and initial != data

Then we just need to override that field on our form definition:

class ConfigForm(forms.ModelForm):
    password = WriteOnlyField()

    class Meta:
        model = ExternalSystem
        fields = ('username', 'password', 'url')

Please note that this technique should not be used in the situation where you don’t need the user to be able to change a value, but instead just want to render the value. In that case, please omit the field from the form, and just use {{ form.instance.<field_name> }} instead - you can even put that in a disabled text input widget if you really want it to look like the other fields.

I also use a JavaScript affordance on all password fields that default to hiding the value, but allows clicking on a control to toggle the visibility of the value: UIkit Password Field.

Preventing Model Overwrites in Django and Postgres

2020-09-23 @ 19:39:15

here.

I had an idea tonight while helping someone in #django. It revolved around using a postgres trigger to prevent overwrites with stale data.

Consider the following model:

class Person(models.Model):
    first_name = models.TextField()
    last_name = models.TextField()

If we had two users attempting to update a given instance at around the same time, Django would fetch whatever it had in the database when they did the GET request to fetch the form, and display that to them. It would also use whatever they sent back to save the object. In that case, the last update wins. Sometimes, this is what is required, but it does mean that one user’s changes would be completely overwritten, even if they had only changed something that the subsequent user did not change.

There are a couple of solutions to this problem. One is to use something like django-model-utils FieldTracker to record which fields have been changed, and only write those back using instance.save(update_fields=...). If you are using a django Form (and you probably should be), then you can also inspect form.changed_data to see what fields have changed.

However, that may not always be the best behaviour. Another solution would be to refuse to save something that had changed since they initially fetched the object, and instead show them the changes, allow them to update to whatever it should be now, and then resubmit. After which time, someone else may have made changes, but then the process repeats.

But how can we know that the object has changed?

One solution could be to use a trigger (and an extra column).

class Person(models.Model):
    first_name = models.TextField()
    last_name = models.TextField()
    _last_state = models.UUIDField()

And in our database trigger:

CREATE EXTENSION "uuid-ossp";

CREATE OR REPLACE FUNCTION prevent_clobbering()
RETURNS TRIGGER AS $prevent_clobbering$

BEGIN
  IF NEW._last_state != OLD._last_state THEN
    RAISE EXCEPTION 'Object was changed';
  END IF;
  NEW._last_state = uuid_generate_v4();
  RETURN NEW;
END;

$prevent_clobbering$
LANGUAGE plpgsql STRICT IMMUTABLE;

CREATE TRIGGER prevent_clobbering
BEFORE UPDATE ON person_person
FOR EACH ROW EXECUTE PROCEDURE prevent_clobbering();

You’d also want to have some level of handling in Django to capture the exception, and re-display the form. You can’t use the form/model validation handling for this, as it needs to happen during the save.

To make this work would also require the _last_state column to have a DEFAULT uuid_generate_v4(), so that newly created rows would get a value.

This is only a prototype at this stage, but does work as a mechanism for preventing overwrites. As usual, there’s probably more work in the application server, and indeed in the UI that would need to be required for displaying stale/updated values.

What this does have going for it is that it’s happening at the database level. There is no way that an update could happen (unless the request coming in happened to guess what the new UUID was going to be).

What about drawbacks? Well, there is a bit more storage in the UUID, and we need to regenerate a new one each time we save a row. We could have something that checks the other rows looking for changes.

Perhaps we could even have the hash of the previous row’s value stored in this field - that way it would not matter that there had been N changes, what matters is the value the user saw before they entered their changes.

Another drawback is that it’s hard-coded to a specific column. We could rewrite the function to allow defining the column when we create the trigger:

CREATE TRIGGER prevent_clobbering
BEFORE UPDATE ON person_person
FOR EACH ROW EXECUTE PROCEDURE prevent_clobbering('_last_state_');

But that requires a bit more work in the function itself:

CREATE OR REPLACE FUNCTION prevent_clobbering()
RETURNS TRIGGER AS $prevent_clobbering$

BEGIN
  IF to_jsonb(NEW)->TG_ARGV[0] != to_jsonb(OLD)->TG_ARGV[0] THEN
    RAISE EXCEPTION 'Object was changed';
  END IF;
  NEW._last_state = uuid_generate_v4();
  RETURN NEW;
END;

$prevent_clobbering$
LANGUAGE plpgsql STRICT IMMUTABLE;

Django properties from expressions, or ComputedField part 2

2020-09-15 @ 14:21:42

here.

I’ve discussed the concept of a ComputedField in the past. On the weekend, a friend pointed me towards SQL Alchemy’s Hybrid Attributes. The main difference here is that in a ComputedField, the calculation is always done in the database. Thus, if a change is made to the model instance (and it is not yet saved), then the ComputedField will not change it’s value. Let’s look at an example from that original post:

class Person(models.Model):
    first_name = models.TextField()
    last_name = models.TextField()
    display_name = ComputedField(
        Concat(F('first_name'), Value(' '), F('last_name')),
        output_field=models.TextField()
    )

We can use this to query, or as an attribute:

Person.objects.filter(display_name__startswith='foo')
Person.objects.first().display_name

But, if we make changes, we don’t see them until we re-query:

person = Person(first_name='Fred', last_name='Jones')
person.display_name  # This is not set

So, it got me thinking. Is it possible to turn a django ORM expression into python code that can execute and have the same output?

And, perhaps the syntax SQL Alchemy uses is nicer?

class Person(models.Model):
    first_name = models.TextField()
    last_name = models.TextField()

    @shared_property
    def display_name(self):
        return Concat(
            F('first_name'),
            Value(' '),
            F('last_name'),
            output_field=models.TextField(),
        )

The advantage to using the decorator approach is that you could have a more complex expression - but perhaps that is actually a disadvantage. It might be nice to ensure that the code can be turned into a python function, after all.

The first step is to get the expression we need to convert to a python function. Writing a python decorator will give us access to the “function” object - we can just call this, as long as it does not refer to self at all, this can be done without an instance:

class shared_property(object):
    def __init__(self, function):
        expression = function(None)

This gives us the expression object. Because this is a python object, we can just look at it directly, and turn that into an AST. Having a class for parsing this makes things a bit simpler. Let’s look at a parser that can handle this expression.

import ast


class Parser:
    def __init__(self, function):
        # Make a copy, in case this expression is used elsewhere, and we change it.
        expression = function(None).copy()
        tree = self.build_expression(expression)
        # Need to turn this into code...
        self.code = compile(tree, mode='eval', filename=function.func_code.co_filename)

    def build_expression(self, expression):
        # Dynamically find the method we need to call to handle this expression.
        return getattr(self, 'handle_{}'.format(expression.__class__.__name__.lower()))(expression)

    def handle_concat(self, concat):
        # A Concat() contains only one source expression: ConcatPair().
        return self.build_expression(*concat.get_source_expressions())

    def handle_concatpair(self, pair):
        left, right = pair.get_source_expressions()
        return ast.BinOp(
            left=self.build_expression(left),
            op=ast.Add(),
            right=self.build_expression(right),
        )

    def handle_f(self, f):
        # Probably some more work here around transforms/lookups...
        # Set this, because without it we get errors. Will have to
        # figure out a better way to handle this later...
        f.contains_aggregate = False
        return ast.Attribute(
            value=ast.Name(id='self'),
            attr=f.name,
        )

    def handle_value(self, value):
        if value.value is None:
            return ast.Name(id='None')

        if isinstance(value.value, (str, unicode)):
            return ast.Str(s=value.value)

        if isinstance(value.value, (int, float)):
            return ast.Num(n=value.value)

        if isinstance(value.value, bool):
            return ast.Name(id=str(value.value))

        # ... others?
        raise ValueError('Unable to handle {}'.format(value))

There’s a bit more “noise” required in there (every node must have a ctx, and a filename, lineno and col_offset), but they make it a bit harder to follow.

So, we have our expression, and we have turned that into an equivalent python expression, and compiled it…except it won’t compile. We need to wrap it in an ast.Expression(), and then we can compile it (and call it).

Roughly, we’ll end up with a code object that does:

self.first_name + (' ' + self.last_name)

We can call this with our context set:

eval(code, {'self': instance})

But, before we head down that route (I did, but you don’t need to), it’s worth noticing that not all ORM expressions can be mapped directly onto a single python expression. For instance, if we added an optional preferred_name field to our model, our display_name expression may look like:

@shared_property
def display_name(self):
    return Case(
        When(preferred_name__isnull=True, then=Concat(F('first_name'), Value(' '), F('last_name'))),
        When(preferred_name__exact=Value(''), then=Concat(F('first_name'), Value(' '), F('last_name'))),
        default=Concat(F('first_name'), Value(' ('), F('preferred_name'), Value(') ') F('last_name')),
        output_field=models.TextField()
    )

Since this will roughly translate to:

@property
  def display_name(self):
      if all([self.preferred_name is None]):
          return self.first_name + ' ' + self.last_name
      elif all([self.preferred_name == '']):
          return self.first_name + ' ' + self.last_name
      else:
          return self.first_name + ' (' + self.preferred_name + ') ' + self.last_name

Whilst this is still a single ast node, it is not an expression (and cannot easily be turned into an expression - although in this case we could use a dict lookup based on self.preferred_name, but that’s not always going to work). Instead, we’ll need to change our code to generate a statement that contains a function definition, and then evaluate that to get the function object in the context. Then, we’ll have a callable that we can call with our model instance to get our result.

There are a few hitches along the way though. The first is turning our method into both a private field and a property. That is the relatively straightforward part:

class shared_property:
    def __init__(self, function):
        self.parsed = Parser(function)
        context = {}
        eval(self.parsed.code, context)
        self.callable = context[function.func_code.co_name]

    def __get__(self, instance, cls=None):
        # Magic Descriptor method: this method will be called when this property
        # is accessed on the instance.
        if instance is None:
            return self
        return self.callable(instance)

    def contribute_to_class(self, cls, name, private_only=False):
        # Magic Django method: this is called by django on class instantiaton, and allows
        # us to add our field (and ourself) to the model. Mostly this is the same as
        # a normal Django Field class would do, with the exception of setting concrete
        # to false, and using the output_field instead of ourself.
        field = self.parsed.expression.output_field
        field.set_attributes_from_name(name)
        field.model = cls
        field.concrete = False
        # This next line is important - it's the key to having everything work when querying.
        field.cached_col = ExpressionCol(self.parsed.expression)
        cls._meta.add_field(field, private=True)
        if not getattr(cls, field.attname, None):
            setattr(cls, field.attname, self)

There are a few things to note in that last method.

We use the output_field from the expression as the added field.
We mark this field as a private, non-concrete field. This prevents django from writing it back to the database, but it also means it will not appear in a .values() unless we explicitly ask for it. That’s actually fine, because we want the python property to execute instead of just using the value the database gave us.
The cached_col attribute is used when generating queries - we’ll look more at that now.

When I previously wrote the ComputedField implementation, the place I was not happy was with the get_col() method/the cached_col attribute. Indeed, to get that to work, I needed to use inspect to sniff up the stack to find a query instance to resolve the expression.

This time around though, I took a different approach. I was not able to use the regular resolve_expression path, because fields are assumed not to require access to the query to resolve to a Col expression. Instead, we can delay the resolve until we have something that gives us the query object.

class ExpressionCol:
    contains_aggregate = False
    def __init__(self, expression):
        self.expression = expression
        self.output_field = expression.output_field

    def get_lookup(self, name):
        return self.output_field.get_lookup(name)

    def get_transform(self, name):
        return self.output_field.get_transform(name)

    def as_sql(self, compiler, connection):
        resolved = self.expression.resolve_expression(compiler.query)
        return resolve_expression.as_sql(compiler, connection)

    def get_db_converters(self, connection):
      return self.output_field.get_db_converters(connection) + \
             self.expression.get_db_converters(connection)

This doesn’t need to be a full Expression subclass, because it mostly delegates things to the output field, but when it is turned into SQL, it can resolve the expression before then using that resolved expression to build the SQL.

So, let’s see how this works now (without showing the new Nodes that are handled by the Parser):

Person.objects.filter(display_name__startswith='Bob')

Yeah, that correctly limits the queryset. How about the ability to re-evaluate without a db round trip?

person = Person(first_name='Fred', last_name='Jones')
person.display_name  # -> 'Fred Jones'
person.preferred_name = 'Jonesy'
person.display_name  # -> 'Fred (Jonesy) Jones'

Success!

This project is not done yet: I have improved the Parser (as implied) to support more expressions, but there is still a bit more to go. It did occur to me (but not until I was writing this post) that the ComputedField(expression) version may actually be nicer. As hinted, that requires the value to be an expression, rather than a function call. It would be possible to create a function that references self, for instance, and breaks in all sorts of ways.