Modeling evolving values

Hourglass on rocks — photo by Aron Visuals on Unsplash

This article will incrementally add time-related requirements to the Employment model from last time. We’ll see use-cases arising commonly and how to adapt our APIs to them.

In the previous article, we started digging into the problem of representing intervals of time and their associated operations. We attached an EffectivePeriod to an Employment to materialize the fact that the latter would only be effective during a given period of time. More details on that pattern can be found here.

Lets start with Employment carrying two new pieces of information: a job_title and a salary…

KISS first

Keep it simple stupid they say. We add attributes directly on the Employment “model” and we’re done, isn’t it?

class Employment(Model):
    effective_period: EffectivePeriod
    internal_id: UUID
    user_id: UUID
    company_id: UUID
    salary: MoneyAmount
    job_title: str

An employee of the same company can get a raise. So we would need a way to update the salary:

class Employment(Model):
    ...

    def set_salary(self, new_salary: MoneyAmount):
        self.salary = new_salary

We lost something

Problem: destructive operations

Quickly, Sam from the Billing team comes to us… He needs us to keep the old salary because we rebuild invoicing data continuously to fix “mistakes” retroactively…

To get an Employment’s salary we need some extra information: the When.

# Instead of doing this:
employment.salary

# Sam wants to do this:
employment.salary(at_time=dt)

An employment.salary isn’t necessarily the same within the employment.effective_period. It’s annoying because by updating the salary we’re:

• missing at which date the raise (🤞) occurred, and
• forgetting the salary from before, that’s still relevant for Billing purposes.

Solution: remember past salaries

The evolution of the salary could be put on an timeline.

Illustration of the evolution of the salary over time

When updating a salary, we specify the period where this salary is actually effective.

def set_salary(
    self,
    new_salary: MoneyAmount,
    effective_period: EffectivePeriod,
):
    ...

To implement this, we could delegate this to a Timeline container that would record associations between (i) effective periods and (ii) salaries, and then retrieve them.

class Timeline(Generic[T]):
    def record(self, value: T, effective_period: EffectivePeriod):
        ...

     def fetch(self, at_time: datetime) -> T:
        ...

Then Employment could integrate this relationship with its salary this way:

class Employment(Model):
    ...

    salary_timeline: Timeline[MoneyAmount]

    def set_salary(self, new_salary, effective_period):
        self.salary_timeline.record(new_salary, effective_period)

    def salary(self, at_time):
        return self.salary_timeline.fetch(at_time)

This pattern can be called a temporal property, from Martin Fowler’s blog.

Granularity

Problem: we have many temporal properties

We also have Employment.job_title. We can follow the same reasoning for that one than the one about Employment.salary: wrapping it into Timeline.

class Employment(Model):
    ...

    job_title_timeline: Timeline[str]

    def set_job_title(seld, new_job_title, effective_period):
        self.job_title_timeline.record(new_job_title, effective_period)

    def job_title(self, at_time):
        return self.job_title_timeline.fetch(at_time)

Independent properties deserve their own timeline. The issue is that we want fool-proofs API that are consistent, easy to use, and easy to understand. Here, we would need to pass a time to get the job_title and another time to get the salary, it creates both additional work and room for getting inconsistent data.

Solution: consistent “snapshots”

We can provide an API allowing consumers to get EmploymentSnapshot composed of:

1. continuous properties from the Employment like id, effective_period, user_id, …
2. temporal properties from the properties that are Timeline-d at the same time

@dataclass(frozen=True)
class EmploymentSnapshot:
    # Continuous properties
    effective_period: EffectivePeriod
    internal_id: UUID
    company_id: UUID
    user_id: UUID

    # Timeline properties, valued to None if not found at snapshot_at
    salary: MoneyAmount | None
    job_title: str | None

    # Generic snapshot properties
    snapshot_subject: Employment
    snapshot_at: datetime

class Employment(Model):
    ...

    # Most clients would prefer working with fixed point employments
    # rather than with the whole Timeline thing.
    def get_snapshot(self, at_time: datetime) -> EmploymentSnapshot:
        ...

This pattern can be called a temporal object, still from Martin Fowler’s blog.

Time-traveling

Problem: debugging past decisions

At some point, the system will take decisions based on EmploymentSnapshot... A coworker writes us this:

Hi there, I'm working this on-call requests where I'm trying to understand
why we sent an email to this user with the wrong job title. I don't see
it anywhere. Does it ring a bell?

Charlie is right! Looking at the EmploymentSnapshot matching the time of the email, there the job title is the expected one and not the one of the email.

When passing an effective period to set_job_title, we can pass a period that’s in the past. It’s great to change the past according to new information of to fix mistakes. But, the tradeoff is that retroactive changes don’t play well with debugging.

Solution: consider a knowledge timeline

To access all the information, we can update our API to get the snapshot like it was when the email was sent.

class EmploymentSnapshot:
    ...
    snapshot_knowledge_at: datatime


class Employment(Model):
    ...

    def employment_snapshot_at(
        employment_id: UUID,
        at_time: datetime,
        known_at: datetime | None,  # None means the "most up to date knowledge"
    ) -> EmploymentSnapshot:
        ...

We can push down the knowledge timeline down to our Timeline container’s API:

class Timeline(Generic[T]):
    def fetch(self, at_time: datetime, known_at: datetime | None) -> T:
        ...

Alternative approaches to solve this problem are many. Two of them are worth mentioning:

• Event sourcing pattern/architecture (documentation)
• Generic audit log at the database/ORM (example)

Implementation

Finally, we’re getting to it!

You’ll find a Python implementation for the Timeline in this gist.

Conclusion

Those patterns are useful when dealing with an evolving state as soon as we care about what happened. We can introduce them gradually with minor changes to APIs and to their consumers.

This write-up left many questions unaddressed but one in particular: “how does this play with storage and querying?”. I’d be happy to write a follow-up article covering that!