Python Great Features

Duck Typing, ABCs, and Protocols

If it looks like a duck, swims like a duck, and quacks like a duck, then it probably is a duck.

The idea behind duck typing is that you don’t need to check the type of an object before using it. Instead you care about what methods and properties the object has (API). If it has the methods and properties you need, you can use it.

Abstract Base Classes (ABCs) and Protocols enable us to keep the flexibility of duck typing, but also have some kind of type checking.

Abstract Base Classes (ABCs)

ABCs provide a way to enforce a contract between classes.

from abc import ABC, abstractmethod

class Animal(ABC):
    def __init__(self, name):
        self.name = name

    @abstractmethod
    def speak(self):
        raise NotImplementedError("speak() must be implemented")

class Dog(Animal):
    def speak(self):
        return 'Woof!'

dog = Dog("Rex")
print(dog.speak())

Protocols

The term protocols is used for some types supporting structural subtyping.

A structural type defines a set of values not by their class, but by their properties (e.g. attributes, methods, dictionary key/value types)

Protocols are a way to define a set of methods that a class must implement in order to satisfy that protocol during static type checking. Python does not automatically enforce it at runtime. @runtime_checkable permits limited attribute-presence checks, not validation of method signatures.

from typing import Protocol

class Animal(Protocol):
    def speak(self) -> str: ...

class Dog:
    def speak(self) -> str:
        return 'Woof!'

def speak(pet: Animal) -> str:
    return pet.speak()

print(speak(Dog()))  # Passes static type check

Type Hints

Type hints are a way to add type information to your code. They are not enforced by Python, but they can be used by tools like MyPy to check your code for type errors.

def add(x: int, y: int) -> int:
    return x + y

print(add(3, 5))

I will use type hints in the examples I provide in this post.

Functional Features

Closures

Functions can read or mutate state through globals, arguments, or enclosing scopes. A closure retains access to bindings in an enclosing function after that function returns; this can preserve state between calls.

A closure is a function object that remembers values in enclosing scopes. An outer function can return such a function, which continues to refer to its enclosing bindings after the outer call finishes. It means variables in the enclosing scope are kept in memory even after the function has finished executing.

from collections.abc import Callable

def request_token() -> str:
    return "my-secret-token"

def get_token() -> Callable[[], str]:
    _token: str | None = None
    def _get():
        nonlocal _token
        if _token:
            print('Using cached token')
            return _token

        print('Fetching new token')
        # call some API to get the token
        _token = request_token()
        return _token
    return _get

token = get_token()
print(token())
print(token())
print(token())

Fetching new token
my-secret-token
Using cached token
my-secret-token
Using cached token
my-secret-token

The next example uses a mock HTTP client to illustrate token caching. A real client also needs token expiry handling, failed-response handling, and synchronization if concurrent callers may refresh the same token. The examples with top-level await run in an async-capable notebook or the page runner; in a regular script, put the async work inside main() and use asyncio.run(main()).

import asyncio
import base64
from collections.abc import (
    Callable,
    Awaitable,
)
import os

# Mock for browser execution (aiohttp requires server environment)
class _MockResponse:
    async def json(self):
        return {"token": "mock-jwt-token"}
    async def __aenter__(self):
        return self
    async def __aexit__(self, *a):
        pass

class ClientSession:
    def __init__(self, base_url=""):
        pass
    def post(self, url, **kw):
        return _MockResponse()
    async def __aenter__(self):
        return self
    async def __aexit__(self, *a):
        pass

COMMON_HEADERS = {"Content-Type": "application/json"}
os.environ["PUBLIC_KEY"] = "test_public_key"
os.environ["SECRET_KEY"] = "test_secret_key"


def _get_access_token() -> Callable[[ClientSession], Awaitable[str]]:
    current_token: str | None = None

    async def _access_token(session: ClientSession) -> str:
        nonlocal current_token

        if current_token:
            print('Using cached token')
            return current_token

        print('Fetching new token')
        PUBLIC_KEY = os.getenv("PUBLIC_KEY")
        SECRET_KEY = os.getenv("SECRET_KEY")
        text = f"{PUBLIC_KEY}:{SECRET_KEY}"
        encode = base64.b64encode(text.encode("utf-8"))
        token = str(encode, "utf-8")
        authorization = f"Basic {token}"
        headers = {**COMMON_HEADERS, "Authorization": authorization}
        async with session.post(
            "/login",
            headers=headers,
        ) as response:
            content = await response.json()
            if content and content.get("token"):
                current_token = content["token"]

        if current_token is None:
            raise ValueError("Token not found")

        return current_token

    return _access_token


_get_access_token = _get_access_token()

async def main() -> None:
    async with ClientSession("https://api.example.com/") as session:
        token = await _get_access_token(session)
        token = await _get_access_token(session)
        token = await _get_access_token(session)
        token = await _get_access_token(session)

await main()

Fetching new token
Using cached token
Using cached token
Using cached token

Decorators (Higher order functions)

Decorators are a way to wrap a function and add extra functionalities to such a function.

In the following example we define a decorator that adds an access token to the headers of a request. Note that the decorator receives a function that receives a ClientSession instance as first argument, that instance will be used to get the access token without having to open a new session for each request.

from collections.abc import (
    Awaitable,
)
from functools import wraps
from typing import (
    Callable,
    Concatenate,
    ParamSpec,
    TypeVar,
)

# Mock for browser execution (aiohttp requires server environment)
class _MockResponse:
    def __init__(self, data):
        self._data = data
    async def json(self):
        return self._data
    async def __aenter__(self):
        return self
    async def __aexit__(self, *a):
        pass

class ClientSession:
    def __init__(self, base_url=""):
        pass
    def get(self, url, **kw):
        return _MockResponse({"id": 1, "name": "Alice"})
    async def __aenter__(self):
        return self
    async def __aexit__(self, *a):
        pass

async def _get_access_token(session):
    return "mock-token"

T = TypeVar('T')
P = ParamSpec('P')


def add_headers(
    func: Callable[Concatenate[ClientSession, P], Awaitable[T]],
) -> Callable[Concatenate[ClientSession, P], Awaitable[T]]:
    @wraps(func)
    async def decorated(
        session: ClientSession, /, *args: P.args, **kwargs: P.kwargs
    ) -> T:
        token = await _get_access_token(session)
        headers = {
            "Content-Type": "application/json",
            "Accept": "application/json",
            "Authorization": f"Bearer {token}",
        }
        kwargs["headers"] = headers

        return await func(session, *args, **kwargs)

    return decorated


@add_headers
async def get_user(
    session: ClientSession,
    user_id: int,
    *,
    headers: dict[str, str] | None = None
) -> dict[str, object]:
    async with session.get(f"users/{user_id}", headers=headers) as resp:
        return await resp.json()

async with ClientSession("https://api.example.com/") as session:
    user = await get_user(session, 1) # headers will be added by the decorator
    print(user)

wraps

without @wraps the name of the function will be wrapper_function

def mock_decorator(original_function):
    def wrapper_function(*args, **kwargs):
        return original_function(*args, **kwargs)
    return wrapper_function

@mock_decorator
def f(x):
    '''Docstring'''
    return x ** 2

print(f.__name__)
print(f.__doc__)

wrapper_function
None

from functools import wraps

def mock_decorator(original_function):
    @wraps(original_function)
    def wrapper_function(*args, **kwargs):
        return original_function(*args, **kwargs)
    return wrapper_function

@mock_decorator
def f(x):
    '''Docstring'''
    return x ** 2

print(f.__name__)
print(f.__doc__)

f
Docstring

def stateful_function(func):
    cache = {}
    def wrapper_function(*args, **kwargs):
        key = (args, tuple(sorted(kwargs.items())))
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    return wrapper_function

@stateful_function
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(10))

The manual cache above requires hashable arguments and keyword values; it is only a teaching example. Stringifying arguments is not a reliable cache key, because different objects can have the same string representation. For pure functions, prefer functools.cache or lru_cache.

cache

import time
from functools import cache

def timer(function):
    def wrapper(*args, **kwargs):
        start = time.time()
        result = function(*args, **kwargs)
        end = time.time()
        print(f'Elapsed time: {end - start}')
        return result
    return wrapper

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

@timer
def fibonacci_timer(n):
    return fibonacci(n)

print(fibonacci_timer(10))

This is kinda like a cache (memoization), so that the function doesn’t have to be called again if the arguments are the same.

Working with Files

Generators

Generators are a special class of functions that simplify the task of writing iterators. Generators are a simple and powerful tool for creating iterators. They are written like regular functions but use the yield statement whenever they want to return data. Each time next() is called on it, the generator resumes where it left off (it remembers all the data values and which statement was last executed).

Generators are iterators, a kind of iterable you can only iterate over once. It’s because they do not store all the values in memory, they generate the values on the fly, that allows processing of large amounts of data efficiently.

# Create a sample file for demonstration
with open('file.txt', 'w') as f:
    f.write('Hello from a large file!\n' * 10)

def read_large_file(filename):
    with open(filename) as file:
        while True:
            chunk = file.read(1000)
            if not chunk:
                break
            yield chunk

for chunk in read_large_file('file.txt'):
    print(chunk)

def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

for i in fibonacci(10):
    print(i)

Context Managers

Context managers are used to allocate and release resources precisely when you want to. They are very useful when you are working with external resources like files, network connections, etc. Context managers are normally implemented using a class that implements the special methods __enter__() and __exit__(). The __enter__() method is invoked when the with statement is encountered. The __exit__() method is invoked at the end of the with block.

class OpenFile:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, traceback):
        self.file.close()

with OpenFile('sample.txt', 'w') as f:
    f.write('Testing')

with open('sample.txt', 'r') as f:
    print(f.read())

import time

class Timer:
    def __init__(self):
        self.start = None
        self.end = None

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

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

    def elapsed_time(self):
        return self.end - self.start

with Timer() as timer:
    print('This should take approximately 2 seconds')
    time.sleep(2)

print('Elapsed time: {}'.format(timer.elapsed_time()))

Note that the __exit__() method can optionally return a Boolean value. If it returns True, any exception raised within the with block is suppressed and execution proceeds as if no exception had occurred. If it returns False, any exception raised within the with block propagates to the surrounding code. A return value of None also leaves the exception unsuppressed.

import threading

lock = threading.Lock()

class LockedContext:
    def __init__(self, lock):
        self.lock = lock

    def __enter__(self):
        print('acquiring lock')
        self.lock.acquire()

    def __exit__(self, exc_type, exc_val, traceback):
        print('releasing lock')
        self.lock.release()

with LockedContext(lock):
    print('Lock acquired')

# The lock is automatically released when the with block ends,
# even if an error occurs

Recent Posts

Listen to this article