Handling multiple errors in Rust iterator adapters
Approaches for handling multiple errors within iterator adapters
Better FastAPI Background Jobs
A more featureful background task runner for Async apps like FastAPI or Discord bots
Lessons Learnt Building FastAPI Application
Recently I was helping a friend of mine (the gracious host of this website), who had an issue where he wanted to manage his DNS Records in cloudflare with some additional functionality not provided by Cloudflare’s API. I used this as a good opportunity to learn more about FastAPI by building a little microservice abstraction layer (Github Link).
Overall working with FastAPI was a delight, and I would thoroughly recommend it to anyone needing to build an API. FastAPI’s documentation website provides really easy to use starting points for so many elements. The auto-generated docs were great for testing during dev, and demonstrating how the thing worked when I was passing it over to my friend. Type hints made the auto-completes in the IDE super fluid which sped me up massively, and caught a number of bugs quickly. It felt so quick to add really complex logic for the JSON validation and be able to rely on the structure of the data you are getting.
While building the app I learnt (or was reminded of) a number of things, so I thought I’d persist the knowledge…
Being Pydantic about application settings
In addition to pydantic
being nicely incorporated into FastAPI to check the format
of JSON payloads, and give you friendly python objects (lovely auto-complete for your
IDE). It’s also awesome for handling any settings you might want to provide
to your application (e.g. via environment variables).
Classes derived from BaseSettings
will try to populate the variables from equivalently named
environment variable when instantiated
from pydantic import BaseModel, BaseSettings
class SomeNestedSettings(BaseModel):
A: str
B: str
class Settings(BaseSettings):
SQLALCHEMY_DATABASE_URL: str = "sqlite:///./data/sqlite.db"
BACKUP_FILE: str = './data/backup.json'
NESTED_SETTINGS: SomeNestedSettings = {'A': 'Default A Value', 'B': 'Default B Value'}
settings = Settings()
You get the same useful format checking capabilities to give some validation to your settings, so your app can fail to startup with bad settings and give some useful stack trace:
export NESTED_SETTINGS='{"A": "A Override"}'
python
>>> from pydantic import BaseModel, BaseSettings
>>>
>>> class SomeNestedSettings(BaseModel):
... A: str
... B: str
...
>>> class Settings(BaseSettings):
... SQLALCHEMY_DATABASE_URL: str = "sqlite:///./data/sqlite.db"
... BACKUP_FILE: str = './data/backup.json'
... NESTED_SETTINGS: SomeNestedSettings = {'A': 'Default A Value', 'B': 'Default B Value'}
...
>>> settings = Settings()
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "pydantic/env_settings.py", line 34, in pydantic.env_settings.BaseSettings.__init__
File "pydantic/main.py", line 362, in pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for Settings
NESTED_SETTINGS -> B
field required (type=value_error.missing)
Here it’s failing because SomeNestedSettings
can only be instantiated with a full specification
(like the default is in Settings
), as it doesn’t specify attribute level defaults
Logging
I’ve spent most of my time working on existing applications or frameworks that handle the logging, so it’s been a while since I’ve actually setup logging on a new python application. I got tripped up by a few things.
I’m very familiar with inserting this boilerplate:
import logging
logger = logging.getLogger(__name__)
at the start of any module I’m adding to integrate with the existing
logging framework, however just doing this in my main.py
wasn’t resulting in the behaviour I was expecting:
- when running the application standalone, INFO logs were not working.
- when running the application via gunicorn/unicorn in the FastAPI docker container, no logs were working
I’d forgotten that one needs to configure the root logger with logging.basicConfig
. This does a few things:
- It can set the root logging level, WARNING is default, so INFO logs are skipped (which fixed problem 1)
- It adds a log handler to actually do something with the logs e.g. print them to standard out (which fixed problem 2)
In main.py
:
import logging
# ...
logging.basicConfig(
format="%(asctime)s - %(process)s - %(name)s - %(lineno)d - %(levelname)s - %(message)s",
level=logging.INFO,
)
logger = logging.getLogger(__name__)
# ...
logger.info('Log this')
Generics in Type Hints
Given I was using FastAPI and one of the benefits of the framework is that so many things have type hints enabled, which in turn means everything autocompletes in the IDE. So I wanted this to continue everywhere. Generally this is pretty easy, however I got a little stuck when I wrote:
def lookup_in_list_of_dns_records(name: str, dns_records: List[DNSRecord]) -> Optional[DNSRecord]:
return next((dns_record for dns_record in dns_records if dns_record.name == name), None)
I had multiple different versions of the DNSRecord class to handle different fields expected in different parts of the
application (talking to the DB, talking to an external API, serialising JSON payloads to my API). I wanted the function
to work for all of them, which it does because they all had the name
attribute, however the type checking was
complaining.
My first thought was to use the tools I was familiar with and just replace DNSRecord
with
Union[DNSRecordX, DNSRecordY, DNSRecordZ]
but this seemed clunky, and would require changes in multiple places
for every new type/child of DNSRecord
I added. Plus the autocomplete was strange because obviously the type I
passed in to the function is the type I will get out, but the typing didn’t indicate that. Even if I put a type
DNSRecordX
in, the autocomplete thought I could get any of DNSRecordX
, DNSRecordY
, or DNSRecordZ
out. So there
must be a better solution.
I had recently been learning about Rust, and in that subject they touched on a similar concept in other languages
“Generics”. So a quick google later with that keyword in my arsenal, and I find python supports the same concept though
typing.TypeVar
where you can define a type T
which is contained to be a set of classes, or a child of a particular
class which was the case I wanted. So now the class I put in is the same that comes out, and it’s appropriately
constrained, so I know the name
attribute will exist:
from typing import List, Optional, TypeVar
T = TypeVar('T', bound=BaseDNSRecord)
def lookup_in_list_of_dns_records(name: str, dns_records: List[T]) -> Optional[T]:
return next((dns_record for dns_record in dns_records if dns_record.name == name), None)