Python: Production Code Snippets

Production-ready Python patterns: async, security, API integrations.

Python JavaScript TypeScript Java C# C++ Go PHP Rust SQL

On this page:

1. Python: Async Web Scraper 2. Python: Retry Decorator with Backoff 3. Python: JWT Token Verification 4. Python: Input Data Sanitization 5. Python: Telegram Notification Client 51. Python: Data Validation with Pydantic 52. Python: Cache with Expiration (TTL) 53. Python: Building a CLI with argparse 54. Python: Transaction Context Manager 55. Python: Multiprocessing for CPU-bound Tasks 56. Python: Structured Logging (JSON) 57. Python: Rate Limiter Decorator (Token Bucket) 58. Python: Mocking APIs in Tests (pytest)

1. Python: Async Web Scraper

Python

Fetching pages in parallel using asyncio, httpx, and BeautifulSoup.

import asyncio
import httpx
from bs4 import BeautifulSoup

class AsyncScraper:
    def __init__(self, urls):
        self.urls = urls

    async def fetch(self, client, url):
        try:
            r = await client.get(url, timeout=10.0, follow_redirects=True)
            if r.status_code == 200:
                soup = BeautifulSoup(r.text, 'html.parser')
                return {"url": url, "title": soup.title.string.strip() if soup.title else ""}
        except Exception as e:
            return {"url": url, "error": str(e)}

    async def run(self):
        async with httpx.AsyncClient() as client:
            return await asyncio.gather(*[self.fetch(client, u) for u in self.urls])

When to use

Use this pattern when you need to fetch data from many URLs at once, and fetching them one by one would take too long.

Common pitfalls

Without a limit on concurrent connections you can overload the target server or your own network link — add a semaphore to cap the number of simultaneous requests.

2. Python: Retry Decorator with Backoff

Python

Automatic retrying of failed network requests with delay.

import time
from functools import wraps

def retry(retries=3, delay=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < retries:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    if attempts == retries: raise e
                    time.sleep(delay)
        return wrapper
    return decorator

When to use

Use it when talking to external APIs or databases, where occasional network errors are normal.

Common pitfalls

A fixed wait time between attempts is often suboptimal — exponential backoff tends to work better in production than a constant delay.

3. Python: JWT Token Verification

Python

Secure decoding and validation of authorization headers (PyJWT).

import jwt
import os

# UWAGA: nigdy nie zapisuj sekretu na stałe w kodzie.
# Wczytaj go ze zmiennej środowiskowej.
SECRET_KEY = os.environ["JWT_SECRET_KEY"]
ALGORITHM = "HS256"

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return {"valid": True, "payload": payload}
    except jwt.ExpiredSignatureError:
        return {"valid": False, "error": "Token wygasł"}
    except jwt.InvalidTokenError:
        return {"valid": False, "error": "Nieprawidłowy token"}

When to use

Essential in any API that authenticates users via tokens — it checks the validity and integrity of the data before it's used.

Common pitfalls

Never keep a secret key in source code — always load it from environment variables, as in the example above.

4. Python: Input Data Sanitization

Python

Protection against XSS and code injection using html.escape.

import html

def sanitize_input(user_input):
    if not isinstance(user_input, str):
        return ""
    return html.escape(user_input.strip())

When to use

Essential anywhere user-supplied data later ends up in HTML — e.g. comments, forms, profile descriptions.

Common pitfalls

HTML sanitization alone doesn't protect against SQL Injection or application-logic-level attacks — it's just one layer of defense.

5. Python: Telegram Notification Client

Python

Sending advanced system status reports via API.

import requests

def send_telegram_message(token, chat_id, message):
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    payload = {"chat_id": chat_id, "text": message, "parse_mode": "Markdown"}
    response = requests.post(url, json=payload)
    return response.json()

When to use

Useful for sending production error alerts, background job statuses, or daily reports directly to your phone.

Common pitfalls

A bot token should never end up in a public repository — keep it in environment variables or a secret manager.

51. Python: Data Validation with Pydantic

Python

Declarative validation and parsing of input data with automatic type conversion.

from pydantic import BaseModel, EmailStr, field_validator

class UserCreate(BaseModel):
    name: str
    email: EmailStr
    age: int

    @field_validator("age")
    @classmethod
    def check_age(cls, v):
        if v < 18:
            raise ValueError("Użytkownik musi mieć co najmniej 18 lat")
        return v

user = UserCreate(name="Anna", email="anna@example.com", age=25)

When to use

Use it to validate data from APIs, forms, or config files — Pydantic automatically converts types and returns readable errors.

Common pitfalls

Pydantic v1 and v2 use different validator syntax (@validator vs @field_validator) — make sure which version of the library you have installed.

52. Python: Cache with Expiration (TTL)

Python

A decorator that caches function results for a set time, unlike the unbounded-lifetime lru_cache.

import time
from functools import wraps

def ttl_cache(seconds=60):
    def decorator(func):
        cache = {}
        @wraps(func)
        def wrapper(*args):
            now = time.time()
            if args in cache and now - cache[args][1] < seconds:
                return cache[args][0]
            result = func(*args)
            cache[args] = (result, now)
            return result
        return wrapper
    return decorator

@ttl_cache(seconds=30)
def get_exchange_rate(currency):
    # kosztowne zapytanie do zewnętrznego API
    return fetch_rate(currency)

When to use

Useful for data that changes periodically (exchange rates, prices), where a standard lru_cache would keep stale data forever.

Common pitfalls

A cache kept in a plain dict grows without bound if arguments vary widely — in production, consider a maximum cache size or the cachetools library.

53. Python: Building a CLI with argparse

Python

Building a command-line tool with subcommands and flags.

import argparse

def main():
    parser = argparse.ArgumentParser(description="Narzędzie do zarządzania zadaniami")
    subparsers = parser.add_subparsers(dest="command")

    add_parser = subparsers.add_parser("add")
    add_parser.add_argument("title", help="Tytuł zadania")
    add_parser.add_argument("--priority", type=int, default=1)

    args = parser.parse_args()
    if args.command == "add":
        print(f"Dodano zadanie: {args.title} (priorytet {args.priority})")

if __name__ == "__main__":
    main()

When to use

Use it when building your own developer tools or admin scripts that offer more than one invocation option.

Common pitfalls

Not validating args.command == None (the user provided no subcommand) causes a silent failure — handle this case and show a help message.

54. Python: Transaction Context Manager

Python

A custom context manager that guarantees commit/rollback regardless of whether an error occurred.

from contextlib import contextmanager

@contextmanager
def db_transaction(connection):
    try:
        yield connection
        connection.commit()
    except Exception:
        connection.rollback()
        raise
    finally:
        connection.close()

with db_transaction(get_connection()) as conn:
    conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")

When to use

Useful anywhere an operation must end in either a commit or a rollback, regardless of whether something went wrong inside the with block.

Common pitfalls

Forgetting to re-raise (a bare 'raise' with no argument) in the except block makes the error disappear silently, and the caller never learns the transaction failed.

55. Python: Multiprocessing for CPU-bound Tasks

Python

Splitting compute-heavy tasks across multiple CPU cores.

from multiprocessing import Pool
import os

def heavy_computation(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    numbers = [10_000_000, 20_000_000, 15_000_000, 25_000_000]
    with Pool(processes=os.cpu_count()) as pool:
        results = pool.map(heavy_computation, numbers)
    print(results)

When to use

Use it for CPU-intensive tasks (computation, image processing) — unlike asyncio, which only helps with I/O operations.

Common pitfalls

Multiprocessing on Windows and in some environments requires an `if __name__ == '__main__':` block — without it, the code can loop when spawning child processes.

56. Python: Structured Logging (JSON)

Python

Configuring a logger that outputs entries in JSON format instead of plain text.

import logging
import json
from datetime import datetime, timezone

class JsonFormatter(logging.Formatter):
    def format(self, record):
        payload = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
        }
        return json.dumps(payload)

logger = logging.getLogger("app")
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

When to use

Makes log analysis easier in tools like ELK or Datadog, where structured JSON is easier to search than plain text.

Common pitfalls

Logging objects that aren't JSON-serializable (e.g. exceptions without str()) will error inside the formatter itself — always convert complex objects to a string before passing them to the logger.

57. Python: Rate Limiter Decorator (Token Bucket)

Python

Limiting how often a function is called, e.g. when talking to an external API.

import time
from functools import wraps

def rate_limit(calls_per_second=1):
    min_interval = 1.0 / calls_per_second
    def decorator(func):
        last_called = [0.0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait = min_interval - elapsed
            if wait > 0:
                time.sleep(wait)
            last_called[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_second=2)
def call_external_api():
    return fetch_data()

When to use

Useful when integrating with external APIs that have per-second rate limits — it protects against 429 Too Many Requests errors.

Common pitfalls

This approach blocks the thread while waiting (time.sleep) — in an async application, an asyncio.sleep-based equivalent works better.

58. Python: Mocking APIs in Tests (pytest)

Python

Testing code that depends on an external API without making real network requests.

import pytest
from unittest.mock import patch

def get_user_data(user_id):
    import requests
    response = requests.get(f"https://api.example.com/users/{user_id}")
    return response.json()

@patch("requests.get")
def test_get_user_data(mock_get):
    mock_get.return_value.json.return_value = {"id": 1, "name": "Anna"}
    result = get_user_data(1)
    assert result["name"] == "Anna"
    mock_get.assert_called_once_with("https://api.example.com/users/1")

When to use

Essential when testing code that integrates with external services — tests must be fast, deterministic, and independent of the real API's availability.

Common pitfalls

Over-mocking means the test only checks that the function called the mock correctly, not that the business logic actually works — balance it with integration tests.