Extensions¶
Extensions let you create reusable, pluggable components that can be hooked into any Ravyn application. Think of them as self-contained modules that add functionality without modifying your core application code.
What You'll Learn¶
- What extensions are and why they're useful
- Creating custom extensions
- Using the Pluggable system
- Hooking extensions into your app
- Extension lifecycle management
Quick Start¶
from ravyn import Ravyn
from ravyn.extensions import Extension, Pluggable
# Create an extension
class DatabaseExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# This runs when the extension is loaded
app.state.db = connect_database()
print("Database extension loaded!")
# Use the extension
app = Ravyn(
extensions={
"database": Pluggable(DatabaseExtension)
}
)
# Access in handlers
@app.get("/users")
async def get_users(request: Request) -> dict:
db = request.app.state.db
users = await db.fetch_all("SELECT * FROM users")
return {"users": users}
Why Use Extensions?¶
Benefits:¶
-
Reusability - Write once, use in multiple apps
-
Modularity - Keep functionality isolated and organized
-
Maintainability - Changes in one place
-
Shareability - Distribute as packages
-
Clean Code - Separate concerns from main app
Perfect For:¶
- Database connections
- Authentication systems
- Logging configurations
- Monitoring integrations
- Custom middleware
- Third-party service integrations
Extension vs ChildRavyn¶
| Feature | Extension | ChildRavyn |
|---|---|---|
| Purpose | Add functionality | Mount sub-application |
| Use Case | Database, auth, logging | API versioning, modules |
| Lifecycle | Runs on app startup | Independent ASGI app |
| Access | Via app.state |
Via routing |
Info
Extensions and ChildRavyn serve different purposes. Use extensions for shared functionality, ChildRavyn for sub-applications.
Creating Extensions¶
Basic Extension¶
from ravyn import Ravyn
from ravyn.extensions import Extension
class MyExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
"""
This method is called when the extension is loaded.
Args:
app: The Ravyn application instance
**kwargs: Additional parameters passed during registration
"""
# Your initialization logic here
app.state.my_service = MyService()
print("MyExtension loaded!")
Extension with Configuration¶
class CacheExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# Get configuration from kwargs
cache_url = kwargs.get("cache_url", "redis://localhost")
ttl = kwargs.get("ttl", 3600)
# Initialize cache
app.state.cache = RedisCache(cache_url, ttl=ttl)
print(f"Cache extension loaded with TTL={ttl}")
The Pluggable System¶
Pluggable wraps your extension and manages its lifecycle.
Using Pluggable¶
from ravyn import Ravyn
from ravyn.extensions import Pluggable
app = Ravyn(
extensions={
"cache": Pluggable(CacheExtension, cache_url="redis://localhost", ttl=7200)
}
)
How It Works¶
- Registration - Extension added to
extensionsdict - Initialization - Ravyn creates Pluggable wrapper
- Execution -
extend()method called on startup - Access - Extension available via
app.extensions
Hooking Extensions¶
Method 1: Via Application Instance (Automatic)¶
from ravyn import Ravyn
from ravyn.extensions import Pluggable
app = Ravyn(
extensions={
"database": Pluggable(DatabaseExtension),
"cache": Pluggable(CacheExtension, ttl=3600),
"auth": Pluggable(AuthExtension)
}
)
# Extensions automatically loaded on startup
# Access via app.extensions
print(app.extensions["database"])
Method 2: Via Settings¶
from ravyn import RavynSettings
class Settings(RavynSettings):
@property
def extensions(self):
return {
"database": Pluggable(DatabaseExtension),
"cache": Pluggable(CacheExtension, ttl=3600)
}
app = Ravyn(settings_module=Settings)
Method 3: Manual Registration¶
from ravyn import Ravyn
app = Ravyn()
# Register extension manually
extension = MyExtension()
app.add_extension("my_ext", extension)
# Or with Pluggable
app.add_extension("my_ext", Pluggable(MyExtension))
Practical Examples¶
Example 1: Database Extension¶
from ravyn import Ravyn
from ravyn.extensions import Extension
import databases
class DatabaseExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# Get database URL from kwargs or environment
db_url = kwargs.get("database_url", "postgresql://localhost/mydb")
# Create database connection
database = databases.Database(db_url)
# Store in app state
app.state.db = database
# Add startup/shutdown handlers
async def connect():
await app.state.db.connect()
print("Database connected")
async def disconnect():
await app.state.db.disconnect()
print("Database disconnected")
app.on_startup.append(connect)
app.on_shutdown.append(disconnect)
# Use the extension
app = Ravyn(
extensions={
"database": Pluggable(
DatabaseExtension,
database_url="postgresql://localhost/mydb"
)
}
)
# Access in handlers
@app.get("/users")
async def get_users(request: Request) -> dict:
users = await request.app.state.db.fetch_all("SELECT * FROM users")
return {"users": users}
Example 2: Logging Extension¶
from ravyn.extensions import Extension
import logging
class LoggingExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# Configure logging
log_level = kwargs.get("log_level", "INFO")
log_format = kwargs.get("log_format", "%(asctime)s - %(message)s")
logging.basicConfig(
level=getattr(logging, log_level),
format=log_format
)
# Store logger in app state
app.state.logger = logging.getLogger("ravyn_app")
print(f"Logging configured at {log_level} level")
# Use it
app = Ravyn(
extensions={
"logging": Pluggable(LoggingExtension, log_level="DEBUG")
}
)
@app.get("/test")
async def test(request: Request) -> dict:
request.app.state.logger.info("Test endpoint called")
return {"status": "ok"}
Example 3: Authentication Extension¶
from ravyn.extensions import Extension
class AuthExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# Get secret key
secret_key = kwargs.get("secret_key", "default-secret")
# Create auth service
from myapp.auth import AuthService
app.state.auth = AuthService(secret_key=secret_key)
# Add authentication middleware
from myapp.middleware import AuthMiddleware
app.user_middleware.append(AuthMiddleware)
print("Authentication extension loaded")
app = Ravyn(
extensions={
"auth": Pluggable(AuthExtension, secret_key="my-secret-key")
}
)
Extension Ordering¶
Sometimes extensions depend on each other. Use ensure_extension() to control order:
class CacheExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# Ensure database extension is loaded first
app.extensions.ensure_extension("database")
# Now we can use the database
db = app.state.db
app.state.cache = Cache(db)
Common Pitfalls & Fixes¶
Pitfall 1: Not Implementing extend()¶
Problem: Missing required extend() method.
# Wrong - missing extend method
class MyExtension(Extension):
pass # No extend() method!
Solution: Implement extend():
# Correct
class MyExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
# Your initialization logic
pass
Pitfall 2: Not Storing in app.state¶
Problem: Can't access extension resources in handlers.
# Wrong - local variable
class DatabaseExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
db = connect_database() # Local variable, lost after extend()
Solution: Store in app.state:
# Correct
class DatabaseExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
app.state.db = connect_database() # Accessible everywhere
Pitfall 3: Using Both Extension and Pluggable¶
Problem: Confusing Extension with Pluggable.
# Wrong - mixing both
app = Ravyn(
extensions={
"db": DatabaseExtension() # Should use Pluggable
}
)
Solution: Use Pluggable wrapper:
# Correct
app = Ravyn(
extensions={
"db": Pluggable(DatabaseExtension)
}
)
Pitfall 4: Extension Order Issues¶
Problem: Extension depends on another that isn't loaded yet.
# Wrong - cache needs database but database loads after
app = Ravyn(
extensions={
"cache": Pluggable(CacheExtension), # Needs database
"database": Pluggable(DatabaseExtension)
}
)
Solution: Use ensure_extension():
# Correct
class CacheExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
app.extensions.ensure_extension("database")
# Now database is guaranteed to be loaded
app.state.cache = Cache(app.state.db)
Best Practices¶
1. Use Descriptive Names¶
# Good
extensions={
"database": Pluggable(DatabaseExtension),
"cache": Pluggable(CacheExtension),
"auth": Pluggable(AuthExtension)
}
# Bad
extensions={
"ext1": Pluggable(DatabaseExtension),
"thing": Pluggable(CacheExtension)
}
2. Accept Configuration via kwargs¶
# Good - configurable
class MyExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
url = kwargs.get("url", "default")
timeout = kwargs.get("timeout", 30)
# Use configuration
3. Clean Up Resources¶
# Good - proper cleanup
class DatabaseExtension(Extension):
def extend(self, app: Ravyn, **kwargs):
app.state.db = connect_database()
async def cleanup():
await app.state.db.disconnect()
app.on_shutdown.append(cleanup)
Standalone Extensions¶
You can create extensions without inheriting from Extension by implementing the protocol:
class MyStandaloneExtension:
def extend(self, app: Ravyn, **kwargs):
# Your logic
app.state.my_service = MyService()
# Works the same way
app = Ravyn(
extensions={
"my_ext": Pluggable(MyStandaloneExtension)
}
)
Next Steps¶
Now that you understand extensions, explore:
- Lifespan Events - Application lifecycle
- Dependencies - Dependency injection
- Middleware - Request processing
- Application Settings - Configuration