Python Advanced Guide

from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self):
        pass

class Circle(Shape):
    def __init__(self, radius):
        self.radius = radius

    def area(self):
        return 3.14159 * self.radius ** 2

print(Circle(5).area())

class PositiveNumber:
    def __set_name__(self, owner, name):
        self.name = name

    def __get__(self, obj, objtype=None):
        return obj.__dict__.get(self.name, 0)

    def __set__(self, obj, value):
        if value < 0:
            raise ValueError("Must be positive")
        obj.__dict__[self.name] = value

class Order:
    total = PositiveNumber()

o = Order()
o.total = 100
print(o.total)

def repeat(times):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(3)
def greet(name):
    print(f"Hi {name}")

greet("Sufyan")

def running_average():
    total = 0
    count = 0
    average = None
    while True:
        value = yield average
        total += value
        count += 1
        average = total / count

avg = running_average()
next(avg)
print(avg.send(10))
print(avg.send(20))
print(avg.send(30))

import asyncio

async def fetch(name, delay):
    await asyncio.sleep(delay)
    print(f"Done: {name}")
    return name

async def main():
    results = await asyncio.gather(
        fetch("A", 2),
        fetch("B", 1),
        fetch("C", 3),
    )
    print(results)

asyncio.run(main())

import asyncio

async def slow_task():
    await asyncio.sleep(5)
    return "done"

async def main():
    try:
        result = await asyncio.wait_for(slow_task(), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("Too slow!")

asyncio.run(main())

import threading
import time

def worker(name, delay):
    time.sleep(delay)
    print(f"{name} finished")

threads = [threading.Thread(target=worker, args=(f"T{i}", i)) for i in range(1, 4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print("All done")

import multiprocessing

def square(n):
    return n * n

with multiprocessing.Pool(4) as pool:
    results = pool.map(square, [1, 2, 3, 4, 5])

print(results)

from concurrent.futures import ThreadPoolExecutor

def double(n):
    return n * 2

with ThreadPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(double, [1, 2, 3, 4]))

print(results)

import functools, time

@functools.lru_cache(maxsize=None)
def fib(n):
    return n if n < 2 else fib(n - 1) + fib(n - 2)

start = time.perf_counter()
print(fib(35))
print(f"Time: {time.perf_counter() - start:.4f}s")

import tracemalloc

tracemalloc.start()

data = [n ** 2 for n in range(100_000)]

current, peak = tracemalloc.get_traced_memory()
print(f"Current: {current / 1024:.0f} KB")
print(f"Peak: {peak / 1024:.0f} KB")

import cProfile

def slow():
    total = 0
    for i in range(1_000_000):
        total += i
    return total

cProfile.run('slow()')

from typing import Protocol

class Named(Protocol):
    name: str

def greet(item: Named) -> str:
    return f"Hello, {item.name}"

class User:
    def __init__(self, name):
        self.name = name

print(greet(User("Sufyan")))

# pyproject.toml (example)
# [build-system]
# requires = ["setuptools>=68"]
# build-backend = "setuptools.build_meta"
#
# [project]
# name = "mypackage"
# version = "0.1.0"
# description = "A sample package"

print("Project structure:")
print("mypackage/")
print("  mypackage/")
print("    __init__.py")
print("    core.py")
print("  pyproject.toml")
print("  README.md")

# Build commands (run in terminal):
# python -m build
# python -m twine upload dist/*

import sys
print("Release checklist:")
print("1. Update version in pyproject.toml")
print("2. Build: python -m build")
print("3. Upload: python -m twine upload dist/*")
print("4. Verify: pip install yourpkg")

import os

os.environ["DATABASE_URL"] = "postgres://localhost/app"

database_url = os.environ.get("DATABASE_URL", "sqlite:///default.db")
print(database_url)

secret = os.environ.get("SECRET_KEY")
print("Secret loaded:", bool(secret))

import secrets

def make_token():
    return secrets.token_urlsafe(16)

print(make_token())
print(make_token())

from dataclasses import dataclass

@dataclass
class EmailSender:
    smtp_host: str

    def send(self, to, subject, body):
        return f"Sent to {to}: {subject}"

class Notifier:
    def __init__(self, sender):
        self.sender = sender

    def notify(self, user, message):
        return self.sender.send(user, "Update", message)

n = Notifier(EmailSender("smtp.example.com"))
print(n.notify("[email protected]", "Welcome!"))

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

logging.info("Service started")
logging.warning("High memory usage")
try:
    result = 10 / 0
except ZeroDivisionError:
    logging.error("Division by zero attempted")

import os
import logging
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

@dataclass
class Config:
    max_results: int = 10

class FakeDB:
    def query(self, limit):
        return [f"row-{i}" for i in range(limit)]

class Service:
    def __init__(self, db, config):
        self.db = db
        self.config = config

    def search(self):
        logging.info("Running search")
        try:
            return self.db.query(self.config.max_results)
        except Exception as exc:
            logging.error(f"Search failed: {exc}")
            return []

service = Service(FakeDB(), Config(max_results=5))
print(service.search())