0
# Plugin System
1
2
Extensible plugin architecture for integrating with external libraries and extending framework functionality. Litestar's plugin system includes core plugins and protocols for custom plugin development.
3
4
## Capabilities
5
6
### Plugin Protocols
7
8
Base protocols that define the plugin interface and lifecycle hooks.
9
10
```python { .api }
11
class PluginProtocol(Protocol):
12
"""Base protocol for all plugins."""
13
14
def on_app_init(self, app_config: AppConfig) -> AppConfig:
15
"""
16
Hook called during application initialization.
17
18
Parameters:
19
- app_config: Application configuration
20
21
Returns:
22
Modified application configuration
23
"""
24
25
class InitPluginProtocol(PluginProtocol):
26
"""Protocol for initialization plugins."""
27
28
def on_app_init(self, app_config: AppConfig) -> AppConfig:
29
"""Initialize plugin during app setup."""
30
31
class SerializationPluginProtocol(PluginProtocol):
32
"""Protocol for serialization plugins."""
33
34
def supports_type(self, field_definition: FieldDefinition) -> bool:
35
"""
36
Check if plugin supports serializing the given type.
37
38
Parameters:
39
- field_definition: Field definition to check
40
41
Returns:
42
True if plugin can handle the type
43
"""
44
45
def create_dto_for_type(
46
self,
47
field_definition: FieldDefinition,
48
handler_id: str,
49
handler: BaseRouteHandler,
50
) -> type[AbstractDTO]:
51
"""
52
Create DTO for the given type.
53
54
Parameters:
55
- field_definition: Field definition to create DTO for
56
- handler_id: Unique handler identifier
57
- handler: Route handler instance
58
59
Returns:
60
DTO class for the type
61
"""
62
63
class OpenAPISchemaPluginProtocol(PluginProtocol):
64
"""Protocol for OpenAPI schema generation plugins."""
65
66
def is_plugin_supported_type(self, value: Any) -> bool:
67
"""Check if plugin supports the given type for schema generation."""
68
69
def to_openapi_schema(
70
self,
71
field_definition: FieldDefinition,
72
handler_id: str,
73
handler: BaseRouteHandler,
74
) -> Schema:
75
"""
76
Generate OpenAPI schema for the given type.
77
78
Parameters:
79
- field_definition: Field definition
80
- handler_id: Handler identifier
81
- handler: Route handler
82
83
Returns:
84
OpenAPI schema object
85
"""
86
87
class CLIPluginProtocol(PluginProtocol):
88
"""Protocol for CLI plugins."""
89
90
def on_cli_init(self, cli: Group) -> None:
91
"""
92
Hook called during CLI initialization.
93
94
Parameters:
95
- cli: Click CLI group to extend
96
"""
97
```
98
99
### Core Plugin Classes
100
101
Built-in plugin implementations for common functionality.
102
103
```python { .api }
104
class InitPlugin:
105
def __init__(self, config: InitPluginConfig):
106
"""
107
Initialization plugin for app setup.
108
109
Parameters:
110
- config: Plugin configuration
111
"""
112
113
def on_app_init(self, app_config: AppConfig) -> AppConfig:
114
"""Initialize plugin during app setup."""
115
116
class SerializationPlugin:
117
def __init__(self):
118
"""Base serialization plugin."""
119
120
def supports_type(self, field_definition: FieldDefinition) -> bool:
121
"""Check if plugin supports the field type."""
122
123
def create_dto_for_type(
124
self,
125
field_definition: FieldDefinition,
126
handler_id: str,
127
handler: BaseRouteHandler,
128
) -> type[AbstractDTO]:
129
"""Create DTO for the field type."""
130
131
class OpenAPISchemaPlugin:
132
def __init__(self):
133
"""Base OpenAPI schema plugin."""
134
135
def is_plugin_supported_type(self, value: Any) -> bool:
136
"""Check if plugin supports the type."""
137
138
def to_openapi_schema(
139
self,
140
field_definition: FieldDefinition,
141
handler_id: str,
142
handler: BaseRouteHandler,
143
) -> Schema:
144
"""Generate OpenAPI schema."""
145
146
class CLIPlugin:
147
def __init__(self, name: str):
148
"""
149
CLI extension plugin.
150
151
Parameters:
152
- name: Plugin name
153
"""
154
155
def on_cli_init(self, cli: Group) -> None:
156
"""Extend CLI with plugin commands."""
157
158
class DIPlugin:
159
"""Dependency injection plugin."""
160
161
def has_typed_init(self, type_: type[Any]) -> bool:
162
"""Check if type has typed constructor."""
163
164
def get_typed_init(self, type_: type[Any]) -> dict[str, Any]:
165
"""Get typed constructor signature."""
166
167
class ReceiveRoutePlugin:
168
"""Plugin for handling route reception."""
169
170
def receive_route(self, route: HTTPRoute | WebSocketRoute) -> None:
171
"""Handle route registration."""
172
```
173
174
### Plugin Registry
175
176
System for managing and coordinating multiple plugins.
177
178
```python { .api }
179
class PluginRegistry:
180
def __init__(self, plugins: Sequence[PluginProtocol] | None = None):
181
"""
182
Plugin registry for managing plugins.
183
184
Parameters:
185
- plugins: Initial list of plugins
186
"""
187
188
def add_plugin(self, plugin: PluginProtocol) -> None:
189
"""Add a plugin to the registry."""
190
191
def remove_plugin(self, plugin: PluginProtocol) -> None:
192
"""Remove a plugin from the registry."""
193
194
def get_plugins_of_type(self, plugin_type: type[T]) -> list[T]:
195
"""Get all plugins of a specific type."""
196
197
def call_plugin_hooks(
198
self,
199
hook_name: str,
200
*args: Any,
201
**kwargs: Any,
202
) -> list[Any]:
203
"""Call a hook on all plugins that support it."""
204
```
205
206
### Built-in Framework Plugins
207
208
Pre-built plugins for popular Python libraries and frameworks.
209
210
```python { .api }
211
# Pydantic Plugin
212
class PydanticPlugin(SerializationPluginProtocol, OpenAPISchemaPluginProtocol):
213
def __init__(self, prefer_alias: bool = False):
214
"""
215
Pydantic integration plugin.
216
217
Parameters:
218
- prefer_alias: Use field aliases in serialization
219
"""
220
221
def supports_type(self, field_definition: FieldDefinition) -> bool:
222
"""Check if type is a Pydantic model."""
223
224
def create_dto_for_type(
225
self,
226
field_definition: FieldDefinition,
227
handler_id: str,
228
handler: BaseRouteHandler,
229
) -> type[PydanticDTO]:
230
"""Create PydanticDTO for model."""
231
232
# HTMX Plugin
233
class HTMXPlugin(InitPluginProtocol):
234
def __init__(self, config: HTMXConfig | None = None):
235
"""
236
HTMX integration plugin.
237
238
Parameters:
239
- config: HTMX configuration
240
"""
241
242
def on_app_init(self, app_config: AppConfig) -> AppConfig:
243
"""Initialize HTMX support."""
244
245
# Prometheus Plugin
246
class PrometheusPlugin(InitPluginProtocol):
247
def __init__(self, config: PrometheusConfig | None = None):
248
"""
249
Prometheus metrics plugin.
250
251
Parameters:
252
- config: Prometheus configuration
253
"""
254
255
def on_app_init(self, app_config: AppConfig) -> AppConfig:
256
"""Initialize metrics collection."""
257
```
258
259
### Plugin Configuration
260
261
Configuration classes for various plugins.
262
263
```python { .api }
264
class InitPluginConfig:
265
def __init__(
266
self,
267
*,
268
connection_lifespan: Sequence[Callable[..., AsyncContextManager[None]]] | None = None,
269
dto: type[AbstractDTO] | None = None,
270
return_dto: type[AbstractDTO] | None = None,
271
signature_namespace: dict[str, Any] | None = None,
272
type_encoders: TypeEncodersMap | None = None,
273
):
274
"""
275
Configuration for initialization plugins.
276
277
Parameters:
278
- connection_lifespan: Connection lifespan managers
279
- dto: Default DTO for requests
280
- return_dto: Default DTO for responses
281
- signature_namespace: Signature inspection namespace
282
- type_encoders: Type encoder mappings
283
"""
284
285
class HTMXConfig:
286
def __init__(
287
self,
288
*,
289
request_class: type[HTMXRequest] | None = None,
290
response_class: type[HTMXTemplate] | None = None,
291
):
292
"""
293
HTMX plugin configuration.
294
295
Parameters:
296
- request_class: Custom HTMX request class
297
- response_class: Custom HTMX response class
298
"""
299
300
class PrometheusConfig:
301
def __init__(
302
self,
303
*,
304
app_name: str = "litestar",
305
prefix: str = "litestar",
306
labels: dict[str, str] | None = None,
307
exclude_paths: set[str] | None = None,
308
exclude_http_methods: set[str] | None = None,
309
group_paths: bool = False,
310
registry: CollectorRegistry | None = None,
311
):
312
"""
313
Prometheus metrics configuration.
314
315
Parameters:
316
- app_name: Application name for metrics
317
- prefix: Metric name prefix
318
- labels: Default labels for metrics
319
- exclude_paths: Paths to exclude from metrics
320
- exclude_http_methods: HTTP methods to exclude
321
- group_paths: Group similar paths together
322
- registry: Custom metrics registry
323
"""
324
```
325
326
## Usage Examples
327
328
### Creating a Custom Plugin
329
330
```python
331
from litestar.plugins import PluginProtocol
332
from litestar.config import AppConfig
333
import logging
334
335
class LoggingPlugin(PluginProtocol):
336
"""Plugin that configures request/response logging."""
337
338
def __init__(self, logger_name: str = "litestar.requests"):
339
self.logger_name = logger_name
340
self.logger = logging.getLogger(logger_name)
341
342
def on_app_init(self, app_config: AppConfig) -> AppConfig:
343
"""Configure logging during app initialization."""
344
345
# Add custom middleware for request logging
346
def logging_middleware(app: ASGIApp) -> ASGIApp:
347
async def middleware(scope: Scope, receive: Receive, send: Send) -> None:
348
if scope["type"] == "http":
349
request = Request(scope, receive)
350
start_time = time.time()
351
352
async def send_wrapper(message: Message) -> None:
353
if message["type"] == "http.response.start":
354
duration = time.time() - start_time
355
self.logger.info(
356
f"{request.method} {request.url.path} - "
357
f"{message['status']} ({duration:.3f}s)"
358
)
359
await send(message)
360
361
await app(scope, receive, send_wrapper)
362
else:
363
await app(scope, receive, send)
364
return middleware
365
366
# Add to middleware stack
367
if app_config.middleware is None:
368
app_config.middleware = []
369
app_config.middleware.append(logging_middleware)
370
371
return app_config
372
373
# Register the plugin
374
app = Litestar(
375
route_handlers=[...],
376
plugins=[LoggingPlugin("myapp.requests")]
377
)
378
```
379
380
### Serialization Plugin
381
382
```python
383
from litestar.plugins import SerializationPluginProtocol
384
from litestar.dto import AbstractDTO
385
import msgspec
386
387
class MsgspecPlugin(SerializationPluginProtocol):
388
"""Plugin for msgspec serialization support."""
389
390
def supports_type(self, field_definition: FieldDefinition) -> bool:
391
"""Check if type is a msgspec Struct."""
392
return (
393
hasattr(field_definition.annotation, "__msgspec_struct__") or
394
(hasattr(field_definition.annotation, "__origin__") and
395
hasattr(field_definition.annotation.__origin__, "__msgspec_struct__"))
396
)
397
398
def create_dto_for_type(
399
self,
400
field_definition: FieldDefinition,
401
handler_id: str,
402
handler: BaseRouteHandler,
403
) -> type[AbstractDTO]:
404
"""Create MsgspecDTO for the struct type."""
405
from litestar.dto import MsgspecDTO
406
407
return MsgspecDTO[field_definition.annotation]
408
409
# Usage with msgspec models
410
@msgspec.defstruct
411
class Product:
412
name: str
413
price: float
414
category: str
415
416
@post("/products")
417
def create_product(data: Product) -> Product:
418
# Plugin automatically handles msgspec serialization
419
return data
420
421
app = Litestar(
422
route_handlers=[create_product],
423
plugins=[MsgspecPlugin()]
424
)
425
```
426
427
### OpenAPI Schema Plugin
428
429
```python
430
from litestar.plugins import OpenAPISchemaPluginProtocol
431
from litestar.openapi.spec import Schema
432
from decimal import Decimal
433
434
class DecimalSchemaPlugin(OpenAPISchemaPluginProtocol):
435
"""Plugin for generating OpenAPI schema for Decimal types."""
436
437
def is_plugin_supported_type(self, value: Any) -> bool:
438
"""Check if value is a Decimal type."""
439
return value is Decimal or (
440
hasattr(value, "__origin__") and value.__origin__ is Decimal
441
)
442
443
def to_openapi_schema(
444
self,
445
field_definition: FieldDefinition,
446
handler_id: str,
447
handler: BaseRouteHandler,
448
) -> Schema:
449
"""Generate schema for Decimal fields."""
450
return Schema(
451
type="string",
452
format="decimal",
453
pattern=r"^\d+(\.\d+)?$",
454
example="99.99",
455
description="Decimal number as string"
456
)
457
458
# Usage
459
@dataclass
460
class Price:
461
amount: Decimal
462
currency: str
463
464
@post("/prices")
465
def create_price(data: Price) -> Price:
466
return data
467
468
app = Litestar(
469
route_handlers=[create_price],
470
plugins=[DecimalSchemaPlugin()],
471
openapi_config=OpenAPIConfig(title="Pricing API")
472
)
473
```
474
475
### CLI Plugin
476
477
```python
478
from litestar.plugins import CLIPluginProtocol
479
import click
480
481
class DatabaseCLIPlugin(CLIPluginProtocol):
482
"""Plugin that adds database management commands."""
483
484
def __init__(self, db_url: str):
485
self.db_url = db_url
486
487
def on_cli_init(self, cli: click.Group) -> None:
488
"""Add database commands to CLI."""
489
490
@cli.group()
491
def db():
492
"""Database management commands."""
493
pass
494
495
@db.command()
496
def migrate():
497
"""Run database migrations."""
498
click.echo(f"Running migrations on {self.db_url}")
499
# Migration logic here
500
501
@db.command()
502
def seed():
503
"""Seed database with initial data."""
504
click.echo("Seeding database...")
505
# Seeding logic here
506
507
@db.command()
508
@click.option("--confirm", is_flag=True, help="Confirm deletion")
509
def reset(confirm: bool):
510
"""Reset database."""
511
if not confirm:
512
click.echo("Use --confirm to reset database")
513
return
514
click.echo("Resetting database...")
515
# Reset logic here
516
517
# Register plugin
518
app = Litestar(
519
route_handlers=[...],
520
plugins=[DatabaseCLIPlugin("postgresql://localhost/myapp")]
521
)
522
523
# CLI commands available:
524
# litestar db migrate
525
# litestar db seed
526
# litestar db reset --confirm
527
```
528
529
### Using Built-in Plugins
530
531
```python
532
from litestar.plugins.pydantic import PydanticPlugin
533
from litestar.plugins.prometheus import PrometheusPlugin, PrometheusConfig
534
from litestar.plugins.htmx import HTMXPlugin, HTMXConfig
535
from pydantic import BaseModel
536
537
class User(BaseModel):
538
name: str
539
email: str
540
age: int
541
542
# Pydantic plugin for automatic serialization
543
pydantic_plugin = PydanticPlugin(prefer_alias=True)
544
545
# Prometheus metrics plugin
546
prometheus_config = PrometheusConfig(
547
app_name="myapp",
548
prefix="myapp",
549
labels={"service": "api", "version": "1.0"},
550
exclude_paths={"/health", "/metrics"}
551
)
552
prometheus_plugin = PrometheusPlugin(prometheus_config)
553
554
# HTMX plugin for server-side rendering
555
htmx_plugin = HTMXPlugin()
556
557
@post("/users")
558
def create_user(data: User) -> User:
559
# Pydantic plugin handles validation and serialization
560
return data
561
562
@get("/metrics")
563
def metrics() -> str:
564
# Prometheus plugin provides metrics endpoint
565
from prometheus_client import generate_latest
566
return generate_latest()
567
568
app = Litestar(
569
route_handlers=[create_user, metrics],
570
plugins=[pydantic_plugin, prometheus_plugin, htmx_plugin]
571
)
572
```
573
574
### Plugin Registry Management
575
576
```python
577
from litestar.plugins import PluginRegistry
578
579
# Create registry with initial plugins
580
registry = PluginRegistry([
581
LoggingPlugin(),
582
PydanticPlugin(),
583
PrometheusPlugin()
584
])
585
586
# Add more plugins dynamically
587
registry.add_plugin(HTMXPlugin())
588
589
# Get plugins of specific types
590
serialization_plugins = registry.get_plugins_of_type(SerializationPluginProtocol)
591
cli_plugins = registry.get_plugins_of_type(CLIPluginProtocol)
592
593
# Call hooks on all plugins
594
results = registry.call_plugin_hooks(
595
"on_app_init",
596
app_config=AppConfig()
597
)
598
599
# Use registry with app
600
app = Litestar(
601
route_handlers=[...],
602
plugins=registry
603
)
604
```
605
606
### Advanced Plugin with Dependency Injection
607
608
```python
609
from litestar.plugins import InitPluginProtocol
610
from litestar.di import Provide
611
612
class DatabasePlugin(InitPluginProtocol):
613
"""Plugin that provides database connection to routes."""
614
615
def __init__(self, connection_string: str):
616
self.connection_string = connection_string
617
self.pool = None
618
619
async def create_pool(self):
620
"""Create database connection pool."""
621
# Create connection pool (pseudo-code)
622
self.pool = await create_pool(self.connection_string)
623
return self.pool
624
625
async def get_connection(self):
626
"""Get database connection from pool."""
627
if not self.pool:
628
await self.create_pool()
629
return await self.pool.acquire()
630
631
def on_app_init(self, app_config: AppConfig) -> AppConfig:
632
"""Register database dependency."""
633
634
# Add dependency provider
635
if app_config.dependencies is None:
636
app_config.dependencies = {}
637
638
app_config.dependencies["db"] = Provide(self.get_connection)
639
640
# Add lifespan handler for cleanup
641
async def database_lifespan():
642
try:
643
yield
644
finally:
645
if self.pool:
646
await self.pool.close()
647
648
if app_config.lifespan is None:
649
app_config.lifespan = []
650
app_config.lifespan.append(database_lifespan)
651
652
return app_config
653
654
# Usage
655
@get("/users")
656
async def get_users(db=Dependency()) -> list[dict]:
657
# Database connection injected by plugin
658
result = await db.fetch("SELECT * FROM users")
659
return [dict(row) for row in result]
660
661
app = Litestar(
662
route_handlers=[get_users],
663
plugins=[DatabasePlugin("postgresql://localhost/myapp")]
664
)
665
```
666
667
## Types
668
669
```python { .api }
670
# Plugin type union
671
PluginType = (
672
PluginProtocol |
673
InitPluginProtocol |
674
SerializationPluginProtocol |
675
OpenAPISchemaPluginProtocol |
676
CLIPluginProtocol
677
)
678
679
# Configuration types
680
AppConfig = Any # From litestar.config.app module
681
FieldDefinition = Any # From litestar._signature module
682
BaseRouteHandler = Any # From litestar.handlers module
683
684
# OpenAPI types
685
Schema = dict[str, Any]
686
687
# CLI types (from click)
688
Group = click.Group
689
Command = click.Command
690
691
# Prometheus types
692
CollectorRegistry = Any # From prometheus_client
693
694
# Type encoders
695
TypeEncodersMap = dict[Any, Callable[[Any], Any]]
696
697
# Generic type variable for plugin types
698
T = TypeVar("T", bound=PluginProtocol)
699
```