Core Server Functionality

from blacksheep import Application
from rodi import Container
from typing import Optional

# Basic application
app = Application()

# Application with configuration
app = Application(
    debug=True,                    # Enable debug mode
    show_error_details=True,       # Show detailed error pages
    services=Container(),          # Custom DI container
    router=None,                   # Custom router (optional)
    mount=None                     # Mount registry (optional)
)

# CORS configuration
cors_strategy = app.use_cors(
    allow_origins="*",
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Content-Type", "Authorization"],
    allow_credentials=True
)

# Named CORS policies
app.add_cors_policy("api", 
    allow_origins=["https://app.example.com"],
    allow_methods=["GET", "POST"]
)

# Authentication setup
auth_strategy = app.use_authentication()
auth_strategy.add_jwt_bearer(
    valid_audiences=["api"],
    authority="https://auth.example.com"
)

# Authorization setup
authz_strategy = app.use_authorization()

# Session management
app.use_sessions(
    secret_key="your-secret-key",
    session_cookie="session",
    session_max_age=3600
)

# Template engines
app.use_templates()

# Controller registration
app.use_controllers()

from pathlib import Path
from blacksheep import DefaultFileOptions

# Basic static file serving
app.serve_files("./static", root_path="/static")

# Advanced static file configuration
app.serve_files(
    source_folder=Path("./static"),
    discovery=True,                    # Allow directory listing
    cache_time=86400,                  # Cache for 24 hours
    extensions={".jpg", ".png", ".css", ".js"},  # Allowed extensions
    root_path="/assets",               # URL prefix
    index_document="index.html",       # Directory index
    fallback_document="404.html",      # Fallback for SPA
    allow_anonymous=True,              # No auth required
    default_file_options=DefaultFileOptions(
        cache_time=3600,
        content_disposition_type="inline"
    )
)

from blacksheep import Application, use_templates
from blacksheep.server.templating import view, view_async
from jinja2 import PackageLoader, Environment

# Setup templates with PackageLoader
app = Application()
loader = PackageLoader("myapp", "templates")
view_function = use_templates(app, loader, enable_async=True)

# Template rendering functions
def template_name(name: str) -> str:
    """Ensures template name has .html extension"""
    if not name.endswith(".html"):
        return name + ".html"
    return name

def render_template(template: Template, *args, **kwargs) -> str:
    """Render template synchronously"""
    return template.render(*args, **kwargs)

async def render_template_async(template: Template, *args, **kwargs) -> str:
    """Render template asynchronously"""
    return await template.render_async(*args, **kwargs)

def view(jinja_environment: Environment, name: str, model: Any = None, **kwargs) -> Response:
    """Returns HTML Response from synchronous template rendering"""
    pass

async def view_async(jinja_environment: Environment, name: str, model: Any = None, **kwargs) -> Response:
    """Returns HTML Response from asynchronous template rendering"""
    pass

# Using templates in route handlers
@app.get("/users/{user_id}")
async def user_profile(user_id: int, jinja: Environment):
    user = await get_user(user_id)
    return await view_async(jinja, "profile", {"user": user})

# Using the view function returned by use_templates
view_function = use_templates(app, PackageLoader("myapp", "templates"))

@app.get("/dashboard")  
async def dashboard():
    data = {"title": "Dashboard", "items": await get_items()}
    return await view_function("dashboard", data)

# Startup events
@app.on_start
async def configure_database():
    """Initialize database connections"""
    await database.connect()

@app.after_start  
async def log_startup():
    """Log after startup complete"""
    print("Application started successfully")

# Shutdown events
@app.on_stop
async def cleanup():
    """Cleanup resources"""
    await database.disconnect()

# Middleware configuration event
@app.on_middlewares_configuration
def configure_middlewares():
    """Configure middleware after all setup"""
    app.middlewares.append(custom_middleware)

from blacksheep.exceptions import HTTPException, BadRequest

# Handle specific HTTP status codes
@app.exception_handler(404)
async def not_found_handler(app: Application, request: Request, exc: HTTPException):
    return Response(404, content=TextContent("Custom 404 page"))

# Handle specific exception types  
@app.exception_handler(ValueError)
async def value_error_handler(app: Application, request: Request, exc: ValueError):
    return Response(400, content=JSONContent({"error": str(exc)}))

# Handle all HTTP exceptions
@app.exception_handler(HTTPException)
async def http_exception_handler(app: Application, request: Request, exc: HTTPException):
    return Response(exc.status, content=JSONContent({
        "error": "HTTP Error",
        "status": exc.status,
        "message": str(exc)
    }))

# Lifespan context managers
@app.lifespan
@asynccontextmanager
async def app_lifespan():
    """Manage application lifecycle"""
    # Startup
    await initialize_resources()
    yield
    # Shutdown  
    await cleanup_resources()

# Manual startup/shutdown
await app.start()    # Start application
await app.stop()     # Stop application

# Application mounting
sub_app = Application()
app.mount("/api/v2", sub_app)  # Mount sub-application

from blacksheep import Route, Router

# Basic route patterns
@app.route("/")                          # Static route
@app.route("/users")                     # Static path
@app.route("/users/{user_id}")           # Parameter capture
@app.route("/posts/{post_id}/comments")  # Multiple segments

# Typed parameters
@app.route("/users/{user_id:int}")       # Integer parameter
@app.route("/files/{file_path:path}")    # Path parameter (allows /)
@app.route("/products/{price:float}")    # Float parameter
@app.route("/items/{uuid:uuid}")         # UUID parameter
@app.route("/docs/{name:str}")           # String parameter (default)

# HTTP method decorators
@app.get("/users")
async def get_users():
    return json([{"id": 1, "name": "Alice"}])

@app.post("/users")  
async def create_user(data: FromJSON[dict]):
    return json({"created": True, "data": data.value})

@app.put("/users/{user_id:int}")
async def update_user(user_id: int, data: FromJSON[dict]):
    return json({"id": user_id, "updated": True})

@app.delete("/users/{user_id:int}")
async def delete_user(user_id: int):
    return json({"id": user_id, "deleted": True})

@app.patch("/users/{user_id:int}")
async def partial_update(user_id: int, data: FromJSON[dict]):
    return json({"id": user_id, "patched": True})

# Multiple methods
@app.route("/users/{user_id:int}", methods=["GET", "PUT"])
async def user_handler(request: Request, user_id: int):
    if request.method == "GET":
        return json({"id": user_id})
    elif request.method == "PUT":
        data = await request.json()
        return json({"id": user_id, "updated": data})

from blacksheep.server.routing import Router, RouteMatch

# Manual router usage
router = Router()
router.add("GET", "/users", get_users_handler)
router.add("POST", "/users", create_user_handler)

# Find matching routes
match = router.get_match("GET", b"/users/123")
if match:
    handler = match.handler
    route_values = match.values  # {"user_id": "123"}
    
# Route sorting (by specificity)
router.sort_routes()

from blacksheep.server.routing import RoutesRegistry, RegisteredRoute

# Route storage without matching
registry = RoutesRegistry()
registry.add("GET", "/users", handler)

# Iterate routes
for route in registry:
    print(f"{route.method} {route.pattern} -> {route.handler}")

# CORS middleware
app.use_cors(
    allow_origins=["https://example.com"],
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Content-Type", "Authorization"],
    allow_credentials=True,
    max_age=3600
)

# Authentication middleware (automatic when using auth)
app.use_authentication()

# Authorization middleware (automatic when using authz)  
app.use_authorization()

# Session middleware (automatic when using sessions)
app.use_sessions("secret-key")

from typing import Callable, Awaitable

# Simple middleware function
async def logging_middleware(request: Request, handler: Callable[[Request], Awaitable[Response]]) -> Response:
    """Log all requests"""
    print(f"{request.method} {request.url.path}")
    start_time = time.time()
    
    response = await handler(request)
    
    duration = time.time() - start_time
    print(f"Response: {response.status} ({duration:.3f}s)")
    
    return response

# Add to application
app.middlewares.append(logging_middleware)

# Class-based middleware
class TimingMiddleware:
    async def __call__(self, request: Request, handler: Callable) -> Response:
        start = time.time()
        response = await handler(request)  
        response.headers.add(b"X-Response-Time", f"{time.time() - start:.3f}".encode())
        return response

app.middlewares.append(TimingMiddleware())

# Middleware execution order (first added = outermost)
app.middlewares.append(cors_middleware)        # 1st - outermost
app.middlewares.append(auth_middleware)        # 2nd
app.middlewares.append(logging_middleware)     # 3rd
app.middlewares.append(timing_middleware)      # 4th - innermost

# Request flow: cors -> auth -> logging -> timing -> handler
# Response flow: handler -> timing -> logging -> auth -> cors

from rodi import Container
from typing import Protocol

# Service interfaces
class DatabaseService(Protocol):
    async def get_user(self, user_id: int) -> dict: ...

class UserRepository:
    def __init__(self, db: DatabaseService):
        self.db = db
        
    async def find_by_id(self, user_id: int) -> dict:
        return await self.db.get_user(user_id)

# Service registration
container = Container()
container.add_singleton(DatabaseService, PostgresDatabase)
container.add_scoped(UserRepository)  # New instance per request
container.add_transient(SomeService)  # New instance each injection

app = Application(services=container)

# Service resolution in handlers
@app.get("/users/{user_id:int}")  
async def get_user(user_id: int, repo: FromServices[UserRepository]):
    user = await repo.find_by_id(user_id)
    return json(user)

# Singleton - single instance for application
container.add_singleton(DatabaseConnection)

# Scoped - single instance per request
container.add_scoped(UserService) 

# Transient - new instance each time
container.add_transient(TemporaryService)

# Instance registration
container.add_instance(config_instance)

# Factory registration  
container.add_factory(lambda: create_complex_service())

# Configure services with settings
@app.on_start
async def configure_services():
    """Configure services at startup"""
    db_service = app.services.get(DatabaseService)
    await db_service.initialize()
    
    cache_service = app.services.get(CacheService) 
    await cache_service.connect()

# Access service provider
service_provider = app.service_provider
user_service = service_provider.get(UserService)

from blacksheep.server.routing import MountRegistry

# Mount FastAPI app
import fastapi
fastapi_app = fastapi.FastAPI()
app.mount("/legacy", fastapi_app)

# Mount static ASGI app
def static_app(scope, receive, send):
    # Custom ASGI application
    pass
    
app.mount("/custom", static_app)

# Mount with registry
mount_registry = MountRegistry(auto_events=True, handle_docs=True)
app = Application(mount=mount_registry)

# Mount registry options
mount_registry = MountRegistry(
    auto_events=True,      # Bind lifecycle events
    handle_docs=False      # Don't handle documentation
)

# Multiple mounts
app.mount("/api/v1", v1_app)
app.mount("/api/v2", v2_app)  
app.mount("/admin", admin_app)
app.mount("/docs", docs_app)

from blacksheep.exceptions import (
    HTTPException, BadRequest, Unauthorized, Forbidden,
    NotFound, InternalServerError, InvalidArgument,
    BadRequestFormat, RangeNotSatisfiable, MessageAborted,
    NotImplementedByServer, InvalidOperation
)

# Raise HTTP exceptions
async def protected_handler():
    if not user.is_authenticated:
        raise Unauthorized("Please log in")
        
    if not user.has_permission("admin"):
        raise Forbidden("Admin access required")
        
    if not user.exists:
        raise NotFound()

# Additional HTTP exceptions
async def advanced_error_handling():
    # Bad request with detailed format error
    try:
        data = parse_invalid_json()
    except ValueError as e:
        raise BadRequestFormat("Invalid JSON format", e)
    
    # Range request handling  
    if not range_satisfiable(request_range):
        raise RangeNotSatisfiable("Requested range not available")
    
    # Feature not implemented
    if feature_not_supported():
        raise NotImplementedByServer("Feature not available")
        
    # Invalid operation
    if operation_not_allowed():
        raise InvalidOperation("Operation not permitted in current state")

# Message aborted during upload
@app.post("/upload")
async def handle_upload(request: Request):
    try:
        content = await request.body()
    except MessageAborted:
        return Response(400, content=TextContent("Upload interrupted"))
        
# Custom HTTP exception
class RateLimitExceeded(HTTPException):
    def __init__(self):
        super().__init__(429, "Rate limit exceeded")

@app.exception_handler(RateLimitExceeded)
async def handle_rate_limit(app, request, exc):
    return Response(429, content=JSONContent({
        "error": "Too many requests",
        "retry_after": 60
    }))

from blacksheep.server.errors import ServerErrorDetailsHandler

# Custom error details
class CustomErrorHandler(ServerErrorDetailsHandler):
    async def handle_error_details(self, request: Request, exc: Exception) -> Response:
        if app.debug:
            return Response(500, content=JSONContent({
                "error": str(exc),
                "type": type(exc).__name__,
                "traceback": traceback.format_exc()
            }))
        return Response(500, content=TextContent("Internal Server Error"))

app.server_error_details_handler = CustomErrorHandler()