Async method decorator

Had a complete headache trying to figure out how a decorator as a class can maintain the possible async properties of a method. The solution is actually very simple. When called, use inspect.iscoroutinefunction to check whether it is a coroutine, and return again an async method! The example adds given paths to a registry, import inspect from functools import wraps paths_registry = [] class route(object): def __init__(self, path: str, **kwargs) -> None: self._path = path def __call__(self, fn): paths_registry.append( { "path": self._path, } ) @wraps(fn) def decorated(*args, **kwargs): return fn(*args, **kwargs) @wraps(fn) async def asyncdecorated(*args, **kwargs): return await fn(*args, **kwargs) if inspect.iscoroutinefunction(fn): return asyncdecorated return decorated A method with this decorator would look like: ...

September 24, 2021 · 1 min · 131 words · Joost

A Simple Factory for Domain Events

This is a simple demonstration of a domain event factory in Python. I assume you are familiar with the Factory Method Pattern. I also use the pydantic package for attribute validation. When implemented, we can use the factory to create immutable domain events with a homogenous data structure across instances of the same type. The metadata is generated by the underlying BaseEvent. In this approach we always produces complete events. Start with all our imports: ...

January 25, 2021 · 2 min · 403 words · Joost

Python Immutable Objects

While reading into implementing DDD, there is often a plea for the use of immutable objects. The main motivation is that an object that is initially valid always remains valid; there is no later verification or validation required. Secondly, working with an immutable object cannot cause any side effects. Some data objects in Python are immutable, the dataclasses themselve are not. Let’s have this simple class: class SimpleClass: def __init__(self, attr1: int): self.attr1 = attr1 With this setup, you can change any of its attributes making the object not immutable: ...

January 13, 2021 · 1 min · 199 words · Joost

How does Airflow schedule Daylight Saving Time?

One morning you find out your favorite Airflow DAG did not ran that night. Sad… Six months later the task ran twice and now you understand: you scheduled your DAG timezone aware and the clock goes back and forth sometimes because of Daylight Saving Time. For example, in Central European Time (CET) on Sunday 29 March 2020, 02:00, the clocks were turned from “local standard time” forward 1 hour to 03:00:00 “local daylight time”. And recently, on Sunday 25 October 2020, 03:00 the clocks were turned from “local daylight time” backward 1 hour to “local standard time” instead. That means that any activity between 02:00 and 03:00 will not exist, or exists twice. ...

October 29, 2020 · 2 min · 326 words · Joost

Control-flow structure for database connections

With Python, creating a database connection is straightforward. Yet, I often see the following case go wrong, while a simple solution is easily at hand by using the context manager pattern. For database connections, you’ll need at least one secret. Let’s say you get this secret from a secret manager by running the get_secret() method. You also use an utility like JayDeBeApi to setup the connection and you are smart enough to close the connection after querying and deleting the password: ...

October 5, 2020 · 2 min · 386 words · Joost