Python Advanced

class Employee:
    company = class="str">"Acme"   class=class="str">"com"># class attribute

    def __init__(self, name: str, salary: float):
        self.name = name
        self.salary = salary

    def raise_pay(self, pct: float) -> None:
        self.salary *= (1 + pct)

    def __repr__(self) -> str:
        return fclass="str">"Employee({self.name!r}, {self.salary})"

class Manager(Employee):
    def __init__(self, name, salary, reports):
        super().__init__(name, salary)
        self.reports = reports

from dataclasses import dataclass

@dataclass
class Point:
    x: float
    y: float

    def distance_from_origin(self) -> float:
        return (self.x ** 2 + self.y ** 2) ** 0.5

p = Point(3, 4)
print(p.distance_from_origin())   class=class="str">"com"># 5.0

import time
from functools import wraps

def timed(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = fn(*args, **kwargs)
        print(fclass="str">"{fn.__name__} took {time.perf_counter() - start:.3f}s")
        return result
    return wrapper

@timed
def slow_query():
    time.sleep(1)
    return class="str">"done"

def read_large_file(path: str):
    with open(path) as f:
        for line in f:
            yield line.rstrip()

class=class="str">"com"># Process millions of rows without loading them all
for line in read_large_file(class="str">"logs.txt"):
    if class="str">"ERROR" in line:
        print(line)

class=class="str">"com"># Generator expression
total = sum(n * n for n in range(1_000_000))

from contextlib import contextmanager

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

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

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def main():
    urls = [class="str">"https://example.com"] * 10
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*(fetch(session, u) for u in urls))
    print(fclass="str">"Fetched {len(results)} pages")

asyncio.run(main())

from typing import Optional

def find_user(user_id: int) -> Optional[dict[str, str]]:
    row = db.fetch_one(class="str">"SELECT * FROM users WHERE id = ?", user_id)
    return dict(row) if row else None

def process(items: list[int]) -> dict[str, int]:
    return {class="str">"total": sum(items), class="str">"count": len(items)}