Context Managers CookBook

# ------------------ errors supressor ----------------

class CatchErrors:

    def __init__(self, errors=Exception):
        self.errors = errors

    def __enter__(self):
        ...

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            return

        return issubclass(exc_type, self.errors)


manager = CatchErrors()
with manager:
    5 / 0  # ZeroDivisionError


manager = CatchErrors(errors=(AttributeError, RuntimeError, ZeroDivisionError))
with manager:
    5 / 0  # ZeroDivisionError

manager = CatchErrors(KeyError)
with manager:
    5 / 0  # ZeroDivisionError




# -------  timer --------------


import time
import random


class Timer:

    start = None
    end = None

    @property
    def total(self):
        return self.end - self.start

    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end = time.perf_counter()


data = [random.random() for _ in range(1_000_000)]
with Timer() as manager:
    sorted(data)


print(f"Total: {manager.total} sec")


# ---------------- access provider -----------

import os


class AccessProvider:

    old = None

    def __enter__(self):
        self.old = os.environ.copy()
        os.environ["USER"] = "test_user"
        os.environ["PASSWORD"] = "P@55w0rd"

    def __exit__(self, exc_type, exc_val, exc_tb):
        os.environ.clear()
        os.environ.update(self.old)


with AccessProvider():
    user = os.environ["USER"]
    password = os.environ["PASSWORD"]
    print(f"Connect to DB using {user=}, {password=}")


os.environ["USER"]
password = os.environ["PASSWORD"]
print(f"Connect to DB using {user=}, {password=}")


# -----------    redirect stdout ---------

# приблизна реалізація
import sys


class RedirectStdout:

    def __init__(self, target):
        self.target = target          # Зберегти таргет

    def __enter__(self):
        sys.stdout = self.target      # Підмінити таргет
        return sys.stdout

    def __exit__(self, exc_type, exc_val, exc_tb):
        sys.stdout = sys.__stdout__   # Повернути оригінальне значення



from contextlib import redirect_stdout
from io import StringIO


with StringIO() as file:
    with redirect_stdout(file):
        print("Hi there")
        print("How are you")

    data = file.getvalue()


print(f"Captured stdout: {data}")



# -----------  catch_errors  generator context manager  --------------

from contextlib import contextmanager


@contextmanager
def catch_errors(errors=Exception):

    print("__enter__ logic")
    try:
        yield "resource"
    except errors:
        pass
    finally:
        print("__exit__ logic")


manager = catch_errors()
with manager:
    5 / 0


manager = catch_errors(errors=(AttributeError, RuntimeError, ZeroDivisionError))
with manager:
    5 / 0


manager = catch_errors(KeyError)
with manager:
    5 / 0


# -------------  throttle http homework   ----------------

# pip install requests

import requests


class ThrottleHttp:
    ...


with ThrottleHttp() as throttle:
    try:
        requests.get("https://swapi.dev/api/people/1/")
    except ConnectionError:
        ...

assert len(throttle.throttled_requests) == 1


res = requests.get("https://swapi.dev/api/people/1/")
assert res.status_code == 200