Core Application

from starlette.applications import Starlette
from starlette.routing import BaseRoute
from starlette.middleware import Middleware
from starlette.types import Lifespan
from typing import Sequence, Mapping, Callable, Any

class Starlette:
    """
    Main application class for creating ASGI applications.
    
    The Starlette application instance manages:
    - Request routing 
    - Middleware stack
    - Exception handling
    - Lifecycle events (startup/shutdown)
    - Application state
    """
    
    def __init__(
        self,
        debug: bool = False,
        routes: Sequence[BaseRoute] | None = None,
        middleware: Sequence[Middleware] | None = None,
        exception_handlers: Mapping[Any, Callable] | None = None,
        on_startup: Sequence[Callable] | None = None,
        on_shutdown: Sequence[Callable] | None = None,
        lifespan: Lifespan[Any] | None = None,
    ) -> None:
        """
        Initialize Starlette application.
        
        Args:
            debug: Enable debug mode with detailed error pages
            routes: List of route definitions
            middleware: List of middleware to apply
            exception_handlers: Custom exception handlers
            on_startup: Functions to run on application startup (deprecated)
            on_shutdown: Functions to run on application shutdown (deprecated)  
            lifespan: Async context manager for application lifespan
        """
    
    # Properties
    @property
    def routes(self) -> list[BaseRoute]:
        """List of application routes."""
    
    @property
    def router(self) -> Router:
        """Router instance handling request dispatch."""
    
    @property 
    def state(self) -> State:
        """Application state object for sharing data."""
    
    # Core methods
    def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
        """Generate URL path for named route."""
    
    def build_middleware_stack(self) -> ASGIApp:
        """Build the middleware stack around the application."""
    
    # Route management
    def mount(
        self, 
        path: str, 
        app: ASGIApp, 
        name: str | None = None
    ) -> None:
        """Mount an ASGI application at the given path."""
    
    def host(
        self, 
        host: str, 
        app: ASGIApp, 
        name: str | None = None
    ) -> None:
        """Add host-based routing for the application."""
    
    def add_route(
        self,
        path: str,
        route: Callable[[Request], Awaitable[Response] | Response],
        methods: list[str] | None = None,
        name: str | None = None,
        include_in_schema: bool = True,
    ) -> None:
        """Add an HTTP route to the application."""
    
    def add_websocket_route(
        self,
        path: str,
        route: Callable[[WebSocket], Awaitable[None]],
        name: str | None = None,
    ) -> None:
        """Add a WebSocket route to the application."""
    
    # Middleware management  
    def add_middleware(
        self,
        middleware_class: type,
        *args: Any,
        **kwargs: Any,
    ) -> None:
        """Add middleware to the application stack."""
    
    # Exception handling
    def add_exception_handler(
        self,
        exc_class_or_status_code: int | type[Exception],
        handler: Callable,
    ) -> None:
        """Add custom exception handler."""
    
    # Event handlers (deprecated - use lifespan instead)
    def add_event_handler(
        self,
        event_type: str,  # "startup" or "shutdown"
        func: Callable,
    ) -> None:
        """Add event handler (deprecated)."""
    
    # Deprecated decorators (will be removed in v1.0.0)
    def exception_handler(
        self, 
        exc_class_or_status_code: int | type[Exception]
    ) -> Callable:
        """Decorator for exception handlers (deprecated)."""
    
    def route(
        self,
        path: str,
        methods: list[str] | None = None,
        name: str | None = None,
        include_in_schema: bool = True,
    ) -> Callable:
        """Decorator for routes (deprecated)."""
    
    def websocket_route(
        self, 
        path: str, 
        name: str | None = None
    ) -> Callable:
        """Decorator for WebSocket routes (deprecated)."""
    
    def middleware(self, middleware_type: str) -> Callable:
        """Decorator for middleware (deprecated)."""

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

async def homepage(request):
    return JSONResponse({"message": "Hello World"})

app = Starlette(
    debug=True,
    routes=[
        Route("/", homepage),
    ]
)

from starlette.applications import Starlette
from starlette.config import Config
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.routing import Route

# Load configuration
config = Config(".env")
DEBUG = config("DEBUG", cast=bool, default=False)

# Configure middleware
middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"]),
]

# Configure exception handlers
def http_exception_handler(request, exc):
    return JSONResponse(
        {"error": exc.detail}, 
        status_code=exc.status_code
    )

exception_handlers = {
    HTTPException: http_exception_handler,
}

# Create application
app = Starlette(
    debug=DEBUG,
    routes=routes,
    middleware=middleware,
    exception_handlers=exception_handlers,
)

# Deprecated approach using on_startup/on_shutdown
async def startup():
    print("Application is starting up...")
    # Initialize database connections, load ML models, etc.

async def shutdown():
    print("Application is shutting down...")
    # Close database connections, cleanup resources, etc.

app = Starlette(
    on_startup=[startup],
    on_shutdown=[shutdown],
    routes=routes,
)

from contextlib import asynccontextmanager
from starlette.applications import Starlette

# Modern approach using lifespan context manager
@asynccontextmanager
async def lifespan(app: Starlette):
    # Startup
    print("Application starting up...")
    # Initialize resources
    app.state.database = await database.connect()
    app.state.ml_models = await load_ml_models()
    
    yield  # Application runs here
    
    # Shutdown
    print("Application shutting down...")
    # Cleanup resources
    await app.state.database.disconnect()
    await cleanup_ml_models(app.state.ml_models)

app = Starlette(
    lifespan=lifespan,
    routes=routes,
)

@asynccontextmanager
async def lifespan(app: Starlette):
    # Startup phase
    try:
        # Initialize database
        app.state.database = await database.connect()
        
        # Initialize cache
        app.state.cache = await cache.connect()
        
        # Load configuration
        app.state.config = await load_application_config()
        
        print("Application startup complete")
        
    except Exception as e:
        print(f"Failed to start application: {e}")
        raise
    
    yield  # Application is running
    
    # Shutdown phase
    try:
        # Close connections gracefully
        if hasattr(app.state, 'database'):
            await app.state.database.disconnect()
        
        if hasattr(app.state, 'cache'):
            await app.state.cache.disconnect()
            
        print("Application shutdown complete")
        
    except Exception as e:
        print(f"Error during shutdown: {e}")

app = Starlette(lifespan=lifespan, routes=routes)

from starlette.datastructures import State

# Application state is automatically created
app = Starlette()

# Set state during lifespan or route handlers
@asynccontextmanager 
async def lifespan(app: Starlette):
    # Set application-level state
    app.state.database = await connect_to_database()
    app.state.config = load_config()
    app.state.cache = {}
    
    yield
    
    await app.state.database.disconnect()

# Access state in route handlers
async def get_users(request):
    database = request.app.state.database
    users = await database.fetch_all("SELECT * FROM users")
    return JSONResponse([dict(user) for user in users])

# Access state in middleware
class DatabaseMiddleware:
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            # Access app state
            database = scope["app"].state.database
            # Add to request scope
            scope["database"] = database
        
        await self.app(scope, receive, send)

from starlette.routing import Route

# Named routes for URL generation
routes = [
    Route("/", homepage, name="home"),
    Route("/users/{user_id}", user_detail, name="user_detail"),
    Route("/api/users/{user_id}/posts/{post_id}", post_detail, name="post_detail"),
]

app = Starlette(routes=routes)

# Generate URLs from application
home_url = app.url_path_for("home")  # "/"
user_url = app.url_path_for("user_detail", user_id=123)  # "/users/123"
post_url = app.url_path_for(
    "post_detail", 
    user_id=123, 
    post_id=456
)  # "/api/users/123/posts/456"

# Generate URLs in request handlers
async def some_endpoint(request):
    # Access url_path_for through request
    user_url = request.url_for("user_detail", user_id=123)
    
    return JSONResponse({
        "user_profile_url": str(user_url),
        "home_url": str(request.url_for("home"))
    })

# Add routes dynamically after application creation
app = Starlette()

# Add individual routes  
app.add_route("/health", health_check, methods=["GET"])
app.add_websocket_route("/ws", websocket_endpoint)

# Mount sub-applications  
from starlette.staticfiles import StaticFiles
app.mount("/static", StaticFiles(directory="static"), name="static")

# Mount API sub-application
api_app = Starlette(routes=api_routes)
app.mount("/api/v1", api_app, name="api_v1")

# Host-based routing
admin_app = Starlette(routes=admin_routes)
app.host("admin.example.com", admin_app, name="admin")

from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse, PlainTextResponse

# Global exception handlers
async def http_exception_handler(request, exc):
    return JSONResponse(
        {"error": exc.detail},
        status_code=exc.status_code,
        headers=exc.headers,
    )

async def validation_exception_handler(request, exc):
    return JSONResponse(
        {"error": "Validation failed", "details": str(exc)},
        status_code=422,
    )

async def generic_exception_handler(request, exc):
    return PlainTextResponse(
        "Internal server error",
        status_code=500,
    )

# Configure exception handlers
exception_handlers = {
    HTTPException: http_exception_handler,
    ValueError: validation_exception_handler,
    500: generic_exception_handler,  # By status code
}

app = Starlette(
    exception_handlers=exception_handlers,
    routes=routes,
)

# Add exception handlers dynamically
app.add_exception_handler(KeyError, key_error_handler)
app.add_exception_handler(404, not_found_handler)

# Enable debug mode for development
app = Starlette(debug=True, routes=routes)

# Debug mode provides:
# - Detailed error pages with stack traces
# - Automatic reloading on code changes (with --reload)
# - More verbose logging

# Control debug mode with environment variables
from starlette.config import Config

config = Config()
DEBUG = config("DEBUG", cast=bool, default=False)

app = Starlette(debug=DEBUG, routes=routes)

# Compose applications from multiple modules
from .api import api_app
from .admin import admin_app
from .auth import auth_app

# Main application
app = Starlette()

# Mount sub-applications
app.mount("/api", api_app, name="api")
app.mount("/admin", admin_app, name="admin") 
app.mount("/auth", auth_app, name="auth")

# Add global middleware
app.add_middleware(CORSMiddleware, allow_origins=["*"])
app.add_middleware(GZipMiddleware, minimum_size=1000)

# Add global routes
app.add_route("/", homepage, name="home")
app.add_route("/health", health_check, name="health")

# Static files
app.mount("/static", StaticFiles(directory="static"), name="static")

from starlette.testclient import TestClient

# Test application with lifespan
def test_application():
    with TestClient(app) as client:
        # Lifespan events are executed
        response = client.get("/")
        assert response.status_code == 200
        
        # Test URL generation
        assert client.app.url_path_for("home") == "/"
        
        # Test application state
        assert hasattr(client.app.state, "database")

# Test application without lifespan
def test_routes_only():
    client = TestClient(app)
    response = client.get("/health")
    assert response.status_code == 200
    client.close()  # Manual cleanup when not using context manager