Example: Multi-threading (I/O-bound)

import threading
import time
import requests

def download_site(url):
    print(f"Starting download: {url}")
    response = requests.get(url) # This is an I/O-bound operation
    print(f"Finished download: {url}, size: {len(response.content)} bytes")

urls = [
    "https://www.example.com",
    "https://www.google.com",
    "https://www.bing.com",
]

# threads = []
# for url in urls:
#     thread = threading.Thread(target=download_site, args=(url,))
#     threads.append(thread)
#     thread.start()

# for thread in threads:
#     thread.join() # Wait for all threads to complete
# print("All downloads complete with threading.")

Example: Multi-processing (CPU-bound)

import multiprocessing
import time

def cpu_bound_task(n):
    print(f"Starting CPU-bound task for {n}...")
    sum_val = 0
    for i in range(n):
        sum_val += i * i # CPU-intensive calculation
    print(f"Finished CPU-bound task for {n}, sum: {sum_val}")
    return sum_val

numbers = [10**7, 10**7, 10**7]

# if __name__ == '__main__': # Required for multiprocessing on Windows/macOS
#     processes = []
#     for num in numbers:
#         process = multiprocessing.Process(target=cpu_bound_task, args=(num,))
#         processes.append(process)
#         process.start()

#     for process in processes:
#         process.join() # Wait for all processes to complete
#     print("All CPU-bound tasks complete with multiprocessing.")

Example: Asyncio (High-concurrency I/O)

import asyncio
import time

async def async_fetch_data(url):
    print(f"Async starting fetch: {url}")
    await asyncio.sleep(1) # Simulate async I/O operation (e.g., network call)
    print(f"Async finished fetch: {url}")
    return f"Data from {url}"

async def main_async():
    tasks = [
        async_fetch_data("http://api.service.com/1"),
        async_fetch_data("http://api.service.com/2"),
        async_fetch_data("http://api.service.com/3"),
    ]
    results = await asyncio.gather(*tasks) # Run coroutines concurrently
    print("All async fetches complete:", results)

# To run:
# if __name__ == '__main__':
#     asyncio.run(main_async())

1. Entities (Domain Layer)

Pure business objects, independent of any framework or database.

from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    id: str
    name: str
    email: str
    created_at: datetime = datetime.now()

    def is_active(self) -> bool:
        # Simple domain logic
        return True # Placeholder for more complex logic

2. Use Cases (Application Layer)

Application-specific business rules. Orchestrates entities and interacts with abstract interfaces (e.g., repositories).

from abc import ABC, abstractmethod
from typing import List, Optional
import uuid

# Abstract interface for user storage (defined in domain/application layer)
class UserRepository(ABC):
    @abstractmethod
    def get_by_id(self, user_id: str) -> Optional[User]:
        pass

    @abstractmethod
    def save(self, user: User) -> None:
        pass

    @abstractmethod
    def get_all(self) -> List[User]:
        pass

class CreateUserUseCase:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    def execute(self, name: str, email: str) -> User:
        # Application-specific business logic
        if not "@" in email:
            raise ValueError("Invalid email format")
            
        user_id = str(uuid.uuid4())
        new_user = User(id=user_id, name=name, email=email)
        self.user_repo.save(new_user)
        return new_user

class GetUsersUseCase:
    def __init__(self, user_repo: UserRepository):
        self.user_repo = user_repo

    def execute(self) -> List[User]:
        return self.user_repo.get_all()

3. Adapters (Infrastructure Layer)

Concrete implementations of interfaces defined in the Use Cases layer. Connects to external services like databases.

# Example: In-memory implementation of UserRepository
class InMemoryUserRepository(UserRepository):
    def __init__(self):
        self._users = {} # Simulates a database

    def get_by_id(self, user_id: str) -> Optional[User]:
        return self._users.get(user_id)

    def save(self, user: User) -> None:
        self._users[user.id] = user

    def get_all(self) -> List[User]:
        return list(self._users.values())

# Example: Conceptual SQLAlchemy implementation (requires SQLAlchemy setup)
# from sqlalchemy import create_engine, Column, String, DateTime
# from sqlalchemy.orm import sessionmaker, declarative_base
# Base = declarative_base()

# class UserTable(Base):
#     __tablename__ = 'users'
#     id = Column(String, primary_key=True)
#     name = Column(String)
#     email = Column(String)
#     created_at = Column(DateTime)

# class SQLAlchemyUserRepository(UserRepository):
#     def __init__(self, session_factory):
#         self.session_factory = session_factory

#     def get_by_id(self, user_id: str) -> Optional[User]:
#         with self.session_factory() as session:
#             user_data = session.query(UserTable).filter_by(id=user_id).first()
#             return User(**user_data.__dict__) if user_data else None

#     def save(self, user: User) -> None:
#         with self.session_factory() as session:
#             user_table = UserTable(**user.__dict__)
#             session.add(user_table)
#             session.commit()

#     def get_all(self) -> List[User]:
#         with self.session_factory() as session:
#             return [User(**u.__dict__) for u in session.query(UserTable).all()]

4. Frameworks & Drivers (External Layer)

The outermost layer. Uses the Use Cases to drive the application, without the Use Cases knowing about the framework.

# Example: Flask Web Application (requires Flask)
# from flask import Flask, request, jsonify
# from dependency_injector.containers import DeclarativeContainer
# from dependency_injector.providers import Singleton, Factory

# # Simple Dependency Injection Container (conceptual)
# class Container(DeclarativeContainer):
#     user_repository = Singleton(InMemoryUserRepository) # Or SQLAlchemyUserRepository
#     create_user_use_case = Factory(CreateUserUseCase, user_repo=user_repository)
#     get_users_use_case = Factory(GetUsersUseCase, user_repo=user_repository)

# app = Flask(__name__)
# container = Container()

# @app.route('/users', methods=['POST'])
# def create_user_endpoint():
#     data = request.get_json()
#     name = data.get('name')
#     email = data.get('email')
#     try:
#         user = container.create_user_use_case().execute(name, email)
#         return jsonify({"id": user.id, "name": user.name, "email": user.email}), 201
#     except ValueError as e:
#         return jsonify({"error": str(e)}), 400

# @app.route('/users', methods=['GET'])
# def get_users_endpoint():
#     users = container.get_users_use_case().execute()
#     return jsonify([{"id": u.id, "name": u.name, "email": u.email} for u in users]), 200

# # To run:
# # if __name__ == '__main__':
# #     app.run(debug=True)