Isolate code blocks from each other?

Aren't we all just constantly re-creating the same bits of code?

This goes beyond boilerplate code. We are adding the same bits of code to every project, the same
git shortcuts, the same logs formatters, the same permissions decorators, ...

Here I've started putting together a personal collection of building blocks.

And I'm starting with: Code that isolates (insulates) code blocks.

Hint: The full function is on my blog post

Example: I'm sending emails to all team members.

If 1 person's email causes a bug, I want the code to skip this person and continue sending emails to others.

Naive - no error handling:

def send_all_emails(users: list[User]):
    for user in users:
        send_report_email_to_user(user)

If we have a list of 10 users, but we encounter an unexpected error
with the report for user num 3, then only the first 2 users will get the email, others will not.

With error handling:

def send_all_emails(users: list[User]):
    for user in users:
        with suppress_and_log_exc():
            # ↑ Will catch any Exception, log it correctly
            # and then continue with the next user 
            send_report_email_to_user(user)

What suppress_and_log_exc does

• it catches some Exception class
• logs it properly
• lets the code continue

So, something like this:

@contextmanager
def suppress_and_log_exc(
    *,
    action_desc: str,
    # ↑ Let's require an identifier. The error msg will be more helpful this way.
):
    try:
        yield
    except Exception as exc:
        logger.error(
            f"Error `{exc.__class__.__name__}` occurred while {action_desc}",
            exc_info=exc,
        )

Possible use cases

This code comes in handy whenever you have a list of actions that are independent of each other.

Like for example: we are triggering various side effects after some event.

Maybe a new user has registered, so, we want to:

• send a Slack high-five to the dev team and also
• create a ticket for the customer success team to contact them and also
• ... .

If any of these side effects fail, the others must still be triggered.

Another example: we have a multi-tenant system and want to trigger one Celery task for each tenant.

If the code for creating a Celery task for customer number 5 has a problem, we still want to create the tasks for customers 6 to 1.000.000.

It would be silly, if our code were to fail at say... sending out the monthly bill, with customer number 5 and then not even try to send it to customer number 6 and 7 and so on.

Adding more settings to the contextmanager

We can make the function more customizable by adding a setting for:

1. the log level - some things are in reality just a warning or an info
2. a map of log levels - a specific log level per exception class
3. the exception class that we want to catch - maybe we just care about EmailSendingException
4. more log data - so we can better understand what went wrong when we see this msg in Sentry
5. an error callback - a function that is called, when the error happens, which can be used for custom error-cleanup
6. .. whatever your heart desires .. 💖

So, here is now the full code.

Fin: All together now

Our Example code could now look like this:

def send_all_emails(users: list[User]):
    for user in users:
        with suppress_and_log_exc(
            action_desc="Sending my very special report email",
            extra={"user": user.id}
        ):
            # ↑ Will catch any Exception, log it correctly and then. 
            send_report_email_to_user(user)

Or it could be crazy complicated like so:

import logging


def send_all_emails(users: list[User]):
    for user in users:
        with suppress_and_log_exc(
            action_desc="Sending my very special report email",
            log_level=logging.WARNING, # <- default log level
            log_level_maps={EmailIsInvalidException: logging.INFO},
            exc_types_to_catch=(ReportException, EmailException,),
            extra={"user": user.id}
        ):
            # ↑ Will catch any Exception, log it correctly and then. 
            send_report_email_to_user(user)