🐍

Python

FastAPI, async programming, Celery, data processing, and AI/ML basics.

FastAPI Essentials

**FastAPI** is a modern, fast web framework for building APIs with Python 3.7+. **Key Features:** - **Type hints** - Automatic validation and documentation - **Async support** - Native async/await - **Dependency Injection** - Clean architecture - **Auto-generated docs** - Swagger/OpenAPI built-in - **Pydantic models** - Data validation

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, EmailStr
from typing import Optional

app = FastAPI()

# Pydantic model for validation
class UserCreate(BaseModel):
    email: EmailStr
    name: str
    age: Optional[int] = None

class UserResponse(BaseModel):
    id: int
    email: str
    name: str

# Dependency injection
async def get_db():
    db = AsyncSession()
    try:
        yield db
    finally:
        await db.close()

@app.post("/users", response_model=UserResponse)
async def create_user(
    user: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    db_user = User(**user.dict())
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

Async Python

**asyncio** is Python's library for writing concurrent code using async/await. **Key Concepts:** - **Coroutines** - async def functions - **Event loop** - Manages async execution - **Tasks** - Wrapped coroutines for concurrent execution - **Gather** - Run multiple coroutines concurrently

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str):
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls: list[str]):
    async with aiohttp.ClientSession() as session:
        # Run concurrently
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

# With semaphore for rate limiting
async def fetch_with_limit(urls: list[str], limit: int = 10):
    semaphore = asyncio.Semaphore(limit)

    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch_url(session, url)

    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

Celery for Background Tasks

**Celery** is a distributed task queue for Python. **Components:** - **Broker** - Message transport (Redis, RabbitMQ) - **Worker** - Processes tasks - **Backend** - Stores results - **Beat** - Scheduler for periodic tasks **Best Practices:** - Keep tasks idempotent - Use proper serialization - Handle retries gracefully - Monitor with Flower

from celery import Celery
from celery.exceptions import MaxRetriesExceededError

app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError,)
)
def process_order(self, order_id: int):
    try:
        order = get_order(order_id)
        # Process order
        send_confirmation_email(order)
        return {"status": "success", "order_id": order_id}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60 * (self.request.retries + 1))

# Chaining tasks
from celery import chain

workflow = chain(
    validate_order.s(order_id),
    process_payment.s(),
    send_confirmation.s()
)
result = workflow.apply_async()

# Periodic tasks with Beat
app.conf.beat_schedule = {
    'cleanup-every-hour': {
        'task': 'tasks.cleanup_old_data',
        'schedule': 3600.0,
    },
}

Type Hints & Pydantic

**Type hints** improve code quality and enable better tooling. **Pydantic** provides data validation using type hints. **Benefits:** - Runtime validation - Automatic serialization - IDE autocomplete - Documentation generation

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

class OrderStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(gt=0)
    price: float = Field(gt=0)

class Order(BaseModel):
    id: Optional[int] = None
    customer_email: str
    items: List[OrderItem]
    status: OrderStatus = OrderStatus.PENDING
    created_at: datetime = Field(default_factory=datetime.utcnow)

    @validator('customer_email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email format')
        return v.lower()

    @property
    def total(self) -> float:
        return sum(item.price * item.quantity for item in self.items)

# Usage
order = Order(
    customer_email="Test@Example.com",
    items=[
        {"product_id": 1, "quantity": 2, "price": 29.99},
        {"product_id": 2, "quantity": 1, "price": 49.99}
    ]
)
print(order.total)  # 109.97
print(order.json())  # Serialized JSON

📝 More Python content coming soon - SQLAlchemy, Testing, ML basics...