Python: Produkcyjne Snippet'y Kodu

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])

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

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"}

import html

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

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()

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)

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)

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()

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")

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)

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)

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()

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")