0
# Core Server Functionality
1

2
This document covers the core server-side components of BlackSheep: Application management, routing, middleware, dependency injection, and static file serving.
3

4
## Application Class
5

6
The `Application` class is the main entry point for BlackSheep web applications, providing configuration, routing, middleware management, and lifecycle control.
7

8
### Basic Application Setup
9

10
```python { .api }
11
from blacksheep import Application
12
from rodi import Container
13
from typing import Optional
14

15
# Basic application
16
app = Application()
17

18
# Application with configuration
19
app = Application(
20
    debug=True,                    # Enable debug mode
21
    show_error_details=True,       # Show detailed error pages
22
    services=Container(),          # Custom DI container
23
    router=None,                   # Custom router (optional)
24
    mount=None                     # Mount registry (optional)
25
)
26
```
27

28
### Application Configuration Methods
29

30
```python { .api }
31
# CORS configuration
32
cors_strategy = app.use_cors(
33
    allow_origins="*",
34
    allow_methods=["GET", "POST", "PUT", "DELETE"],
35
    allow_headers=["Content-Type", "Authorization"],
36
    allow_credentials=True
37
)
38

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

45
# Authentication setup
46
auth_strategy = app.use_authentication()
47
auth_strategy.add_jwt_bearer(
48
    valid_audiences=["api"],
49
    authority="https://auth.example.com"
50
)
51

52
# Authorization setup
53
authz_strategy = app.use_authorization()
54

55
# Session management
56
app.use_sessions(
57
    secret_key="your-secret-key",
58
    session_cookie="session",
59
    session_max_age=3600
60
)
61

62
# Template engines
63
app.use_templates()
64

65
# Controller registration
66
app.use_controllers()
67
```
68

69
### Static File Serving
70

71
```python { .api }
72
from pathlib import Path
73
from blacksheep import DefaultFileOptions
74

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

78
# Advanced static file configuration
79
app.serve_files(
80
    source_folder=Path("./static"),
81
    discovery=True,                    # Allow directory listing
82
    cache_time=86400,                  # Cache for 24 hours
83
    extensions={".jpg", ".png", ".css", ".js"},  # Allowed extensions
84
    root_path="/assets",               # URL prefix
85
    index_document="index.html",       # Directory index
86
    fallback_document="404.html",      # Fallback for SPA
87
    allow_anonymous=True,              # No auth required
88
    default_file_options=DefaultFileOptions(
89
        cache_time=3600,
90
        content_disposition_type="inline"
91
    )
92
)
93
```
94

95
### Template Engine Integration
96

97
BlackSheep provides built-in Jinja2 template engine integration for server-side HTML rendering.
98

99
```python { .api }
100
from blacksheep import Application, use_templates
101
from blacksheep.server.templating import view, view_async
102
from jinja2 import PackageLoader, Environment
103

104
# Setup templates with PackageLoader
105
app = Application()
106
loader = PackageLoader("myapp", "templates")
107
view_function = use_templates(app, loader, enable_async=True)
108

109
# Template rendering functions
110
def template_name(name: str) -> str:
111
    """Ensures template name has .html extension"""
112
    if not name.endswith(".html"):
113
        return name + ".html"
114
    return name
115

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

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

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

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

133
#### Template Usage Examples
134

135
```python
136
# Using templates in route handlers
137
@app.get("/users/{user_id}")
138
async def user_profile(user_id: int, jinja: Environment):
139
    user = await get_user(user_id)
140
    return await view_async(jinja, "profile", {"user": user})
141

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

145
@app.get("/dashboard")  
146
async def dashboard():
147
    data = {"title": "Dashboard", "items": await get_items()}
148
    return await view_function("dashboard", data)
149
```
150

151
### Application Events
152

153
```python { .api }
154
# Startup events
155
@app.on_start
156
async def configure_database():
157
    """Initialize database connections"""
158
    await database.connect()
159

160
@app.after_start  
161
async def log_startup():
162
    """Log after startup complete"""
163
    print("Application started successfully")
164

165
# Shutdown events
166
@app.on_stop
167
async def cleanup():
168
    """Cleanup resources"""
169
    await database.disconnect()
170

171
# Middleware configuration event
172
@app.on_middlewares_configuration
173
def configure_middlewares():
174
    """Configure middleware after all setup"""
175
    app.middlewares.append(custom_middleware)
176
```
177

178
### Exception Handling
179

180
```python { .api }
181
from blacksheep.exceptions import HTTPException, BadRequest
182

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

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

193
# Handle all HTTP exceptions
194
@app.exception_handler(HTTPException)
195
async def http_exception_handler(app: Application, request: Request, exc: HTTPException):
196
    return Response(exc.status, content=JSONContent({
197
        "error": "HTTP Error",
198
        "status": exc.status,
199
        "message": str(exc)
200
    }))
201
```
202

203
### Lifecycle Management
204

205
```python { .api }
206
# Lifespan context managers
207
@app.lifespan
208
@asynccontextmanager
209
async def app_lifespan():
210
    """Manage application lifecycle"""
211
    # Startup
212
    await initialize_resources()
213
    yield
214
    # Shutdown  
215
    await cleanup_resources()
216

217
# Manual startup/shutdown
218
await app.start()    # Start application
219
await app.stop()     # Stop application
220

221
# Application mounting
222
sub_app = Application()
223
app.mount("/api/v2", sub_app)  # Mount sub-application
224
```
225

226
## Routing System
227

228
BlackSheep provides a powerful routing system with pattern matching, parameter extraction, and type conversion.
229

230
### Route Patterns
231

232
```python { .api }
233
from blacksheep import Route, Router
234

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

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

249
### HTTP Method Decorators
250

251
```python { .api }
252
# HTTP method decorators
253
@app.get("/users")
254
async def get_users():
255
    return json([{"id": 1, "name": "Alice"}])
256

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

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

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

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

273
# Multiple methods
274
@app.route("/users/{user_id:int}", methods=["GET", "PUT"])
275
async def user_handler(request: Request, user_id: int):
276
    if request.method == "GET":
277
        return json({"id": user_id})
278
    elif request.method == "PUT":
279
        data = await request.json()
280
        return json({"id": user_id, "updated": data})
281
```
282

283
### Router Class
284

285
```python { .api }
286
from blacksheep.server.routing import Router, RouteMatch
287

288
# Manual router usage
289
router = Router()
290
router.add("GET", "/users", get_users_handler)
291
router.add("POST", "/users", create_user_handler)
292

293
# Find matching routes
294
match = router.get_match("GET", b"/users/123")
295
if match:
296
    handler = match.handler
297
    route_values = match.values  # {"user_id": "123"}
298
    
299
# Route sorting (by specificity)
300
router.sort_routes()
301
```
302

303
### Route Registry
304

305
```python { .api }
306
from blacksheep.server.routing import RoutesRegistry, RegisteredRoute
307

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

312
# Iterate routes
313
for route in registry:
314
    print(f"{route.method} {route.pattern} -> {route.handler}")
315
```
316

317
## Middleware System
318

319
BlackSheep uses a middleware pipeline for request/response processing with both built-in and custom middleware support.
320

321
### Built-in Middleware
322

323
```python { .api }
324
# CORS middleware
325
app.use_cors(
326
    allow_origins=["https://example.com"],
327
    allow_methods=["GET", "POST", "PUT", "DELETE"],
328
    allow_headers=["Content-Type", "Authorization"],
329
    allow_credentials=True,
330
    max_age=3600
331
)
332

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

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

339
# Session middleware (automatic when using sessions)
340
app.use_sessions("secret-key")
341
```
342

343
### Custom Middleware
344

345
```python { .api }
346
from typing import Callable, Awaitable
347

348
# Simple middleware function
349
async def logging_middleware(request: Request, handler: Callable[[Request], Awaitable[Response]]) -> Response:
350
    """Log all requests"""
351
    print(f"{request.method} {request.url.path}")
352
    start_time = time.time()
353
    
354
    response = await handler(request)
355
    
356
    duration = time.time() - start_time
357
    print(f"Response: {response.status} ({duration:.3f}s)")
358
    
359
    return response
360

361
# Add to application
362
app.middlewares.append(logging_middleware)
363

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

372
app.middlewares.append(TimingMiddleware())
373
```
374

375
### Middleware Order
376

377
```python { .api }
378
# Middleware execution order (first added = outermost)
379
app.middlewares.append(cors_middleware)        # 1st - outermost
380
app.middlewares.append(auth_middleware)        # 2nd
381
app.middlewares.append(logging_middleware)     # 3rd
382
app.middlewares.append(timing_middleware)      # 4th - innermost
383

384
# Request flow: cors -> auth -> logging -> timing -> handler
385
# Response flow: handler -> timing -> logging -> auth -> cors
386
```
387

388
## Dependency Injection
389

390
BlackSheep uses the `rodi` library for dependency injection, providing service registration and automatic resolution.
391

392
### Service Registration
393

394
```python { .api }
395
from rodi import Container
396
from typing import Protocol
397

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

402
class UserRepository:
403
    def __init__(self, db: DatabaseService):
404
        self.db = db
405
        
406
    async def find_by_id(self, user_id: int) -> dict:
407
        return await self.db.get_user(user_id)
408

409
# Service registration
410
container = Container()
411
container.add_singleton(DatabaseService, PostgresDatabase)
412
container.add_scoped(UserRepository)  # New instance per request
413
container.add_transient(SomeService)  # New instance each injection
414

415
app = Application(services=container)
416

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

424
### Service Lifetimes
425

426
```python { .api }
427
# Singleton - single instance for application
428
container.add_singleton(DatabaseConnection)
429

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

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

436
# Instance registration
437
container.add_instance(config_instance)
438

439
# Factory registration  
440
container.add_factory(lambda: create_complex_service())
441
```
442

443
### Service Configuration
444

445
```python { .api }
446
# Configure services with settings
447
@app.on_start
448
async def configure_services():
449
    """Configure services at startup"""
450
    db_service = app.services.get(DatabaseService)
451
    await db_service.initialize()
452
    
453
    cache_service = app.services.get(CacheService) 
454
    await cache_service.connect()
455

456
# Access service provider
457
service_provider = app.service_provider
458
user_service = service_provider.get(UserService)
459
```
460

461
## Application Mounting
462

463
Mount other ASGI applications within BlackSheep applications.
464

465
### Basic Mounting
466

467
```python { .api }
468
from blacksheep.server.routing import MountRegistry
469

470
# Mount FastAPI app
471
import fastapi
472
fastapi_app = fastapi.FastAPI()
473
app.mount("/legacy", fastapi_app)
474

475
# Mount static ASGI app
476
def static_app(scope, receive, send):
477
    # Custom ASGI application
478
    pass
479
    
480
app.mount("/custom", static_app)
481

482
# Mount with registry
483
mount_registry = MountRegistry(auto_events=True, handle_docs=True)
484
app = Application(mount=mount_registry)
485
```
486

487
### Mount Configuration
488

489
```python { .api }
490
# Mount registry options
491
mount_registry = MountRegistry(
492
    auto_events=True,      # Bind lifecycle events
493
    handle_docs=False      # Don't handle documentation
494
)
495

496
# Multiple mounts
497
app.mount("/api/v1", v1_app)
498
app.mount("/api/v2", v2_app)  
499
app.mount("/admin", admin_app)
500
app.mount("/docs", docs_app)
501
```
502

503
## Error Handling
504

505
Comprehensive error handling with custom error pages and logging.
506

507
### HTTP Exceptions
508

509
```python { .api }
510
from blacksheep.exceptions import (
511
    HTTPException, BadRequest, Unauthorized, Forbidden,
512
    NotFound, InternalServerError, InvalidArgument,
513
    BadRequestFormat, RangeNotSatisfiable, MessageAborted,
514
    NotImplementedByServer, InvalidOperation
515
)
516

517
# Raise HTTP exceptions
518
async def protected_handler():
519
    if not user.is_authenticated:
520
        raise Unauthorized("Please log in")
521
        
522
    if not user.has_permission("admin"):
523
        raise Forbidden("Admin access required")
524
        
525
    if not user.exists:
526
        raise NotFound()
527

528
# Additional HTTP exceptions
529
async def advanced_error_handling():
530
    # Bad request with detailed format error
531
    try:
532
        data = parse_invalid_json()
533
    except ValueError as e:
534
        raise BadRequestFormat("Invalid JSON format", e)
535
    
536
    # Range request handling  
537
    if not range_satisfiable(request_range):
538
        raise RangeNotSatisfiable("Requested range not available")
539
    
540
    # Feature not implemented
541
    if feature_not_supported():
542
        raise NotImplementedByServer("Feature not available")
543
        
544
    # Invalid operation
545
    if operation_not_allowed():
546
        raise InvalidOperation("Operation not permitted in current state")
547

548
# Message aborted during upload
549
@app.post("/upload")
550
async def handle_upload(request: Request):
551
    try:
552
        content = await request.body()
553
    except MessageAborted:
554
        return Response(400, content=TextContent("Upload interrupted"))
555
        
556
# Custom HTTP exception
557
class RateLimitExceeded(HTTPException):
558
    def __init__(self):
559
        super().__init__(429, "Rate limit exceeded")
560

561
@app.exception_handler(RateLimitExceeded)
562
async def handle_rate_limit(app, request, exc):
563
    return Response(429, content=JSONContent({
564
        "error": "Too many requests",
565
        "retry_after": 60
566
    }))
567
```
568

569
### Error Details Handler
570

571
```python { .api }
572
from blacksheep.server.errors import ServerErrorDetailsHandler
573

574
# Custom error details
575
class CustomErrorHandler(ServerErrorDetailsHandler):
576
    async def handle_error_details(self, request: Request, exc: Exception) -> Response:
577
        if app.debug:
578
            return Response(500, content=JSONContent({
579
                "error": str(exc),
580
                "type": type(exc).__name__,
581
                "traceback": traceback.format_exc()
582
            }))
583
        return Response(500, content=TextContent("Internal Server Error"))
584

585
app.server_error_details_handler = CustomErrorHandler()
586
```
587

588
This comprehensive core server functionality provides the foundation for building scalable web applications with BlackSheep. The framework's modular design allows you to use only the components you need while maintaining high performance and type safety.