Advanced Application Framework for Python with a focus on Command Line Interfaces
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
The interface and handler system provides the foundation for cement's pluggable architecture. Interfaces define contracts for functionality while handlers provide concrete implementations, enabling customizable behavior across all framework components.
Base class for defining framework interfaces that specify contracts for functionality.
class Interface:
"""
Base interface class that all cement interfaces should inherit from.
Interfaces define contracts and specifications for functionality
that handlers must implement. They provide the blueprint for
pluggable components in the cement framework.
"""
def __init__(self, **kw: Any) -> None:
"""
Initialize the interface.
Args:
**kw: Additional keyword arguments
"""
def _validate(self) -> None:
"""
Perform validation to ensure proper interface definition.
This method should be overridden by interface implementations
to validate interface-specific requirements.
"""
@property
def _meta(self) -> Any:
"""Interface meta-data configuration object."""Base class for implementing interface contracts through concrete handler implementations.
class Handler:
"""
Base handler class that all cement handlers should inherit from.
Handlers provide concrete implementations of interface contracts.
They contain the actual functionality that interfaces define.
"""
def __init__(self, **kw: Any) -> None:
"""
Initialize the handler.
Args:
**kw: Additional keyword arguments
"""
def _validate(self) -> None:
"""
Perform validation to ensure proper handler implementation.
This method should be overridden by handler implementations
to validate handler-specific requirements and interface compliance.
"""
@property
def _meta(self) -> Any:
"""Handler meta-data configuration object."""Manages the interface system including interface definition, registration, and retrieval.
class InterfaceManager:
"""
Manages the interface system for defining and retrieving interfaces.
The interface manager provides centralized control over interface
registration and lookup functionality.
"""
def define(self, interface: Interface) -> None:
"""
Define a new interface in the system.
Args:
interface: Interface class to define
"""
def defined(self, interface: str) -> bool:
"""
Check if an interface is defined.
Args:
interface: Interface name to check
Returns:
True if interface is defined, False otherwise
"""
def list(self) -> List[str]:
"""
Get list of all defined interfaces.
Returns:
List of interface names
"""
def get(self, interface: str) -> Interface:
"""
Get an interface by name.
Args:
interface: Interface name to retrieve
Returns:
Interface class
Raises:
InterfaceError: If interface is not found
"""Manages the handler system including handler registration, resolution, and retrieval.
class HandlerManager:
"""
Manages the handler system for registering and resolving handlers.
The handler manager provides centralized control over handler
registration, lookup, and resolution functionality.
"""
def register(self, handler: Handler) -> None:
"""
Register a handler in the system.
Args:
handler: Handler class to register
"""
def registered(self, interface: str, handler: str) -> bool:
"""
Check if a handler is registered for an interface.
Args:
interface: Interface name
handler: Handler name
Returns:
True if handler is registered, False otherwise
"""
def list(self, interface: str) -> List[str]:
"""
Get list of registered handlers for an interface.
Args:
interface: Interface name
Returns:
List of handler names for the interface
"""
def get(self, interface: str, handler: str) -> Handler:
"""
Get a handler instance by interface and handler name.
Args:
interface: Interface name
handler: Handler name
Returns:
Handler instance
Raises:
FrameworkError: If handler is not found
"""
def resolve(self, interface: str, handler: str) -> Handler:
"""
Resolve and return a handler instance.
Args:
interface: Interface name
handler: Handler name
Returns:
Resolved handler instance
Raises:
FrameworkError: If handler cannot be resolved
"""Interface behavior is controlled through the Meta class that defines interface properties.
class Meta:
"""
Interface meta-data configuration.
Controls interface behavior and properties.
"""
interface: str = None
"""The string identifier of this interface"""Handler behavior is controlled through the Meta class that defines handler properties and interface association.
class Meta:
"""
Handler meta-data configuration.
Controls handler behavior, interface association, and configuration.
"""
label: str = None
"""The string identifier of this handler"""
interface: str = None
"""The interface that this handler implements"""
config_section: str = None
"""Configuration section to merge config_defaults with"""
config_defaults: Dict[str, Any] = None
"""Default configuration dictionary"""
overridable: bool = False
"""Whether this handler can be overridden"""from cement import Interface
class DatabaseInterface(Interface):
"""
Interface for database operations.
Defines the contract that database handlers must implement.
"""
class Meta:
interface = 'database'
def connect(self, **kwargs):
"""
Connect to database.
This method must be implemented by handlers.
"""
raise NotImplementedError
def execute(self, query, params=None):
"""
Execute a database query.
Args:
query: SQL query string
params: Query parameters
This method must be implemented by handlers.
"""
raise NotImplementedError
def close(self):
"""
Close database connection.
This method must be implemented by handlers.
"""
raise NotImplementedErrorfrom cement import Handler
import sqlite3
class SQLiteHandler(Handler, DatabaseInterface):
"""
SQLite implementation of the database interface.
"""
class Meta:
label = 'sqlite'
interface = 'database'
config_defaults = {
'database_file': './app.db',
'timeout': 30
}
def __init__(self, **kw):
super().__init__(**kw)
self.connection = None
def connect(self, **kwargs):
"""Connect to SQLite database."""
db_file = kwargs.get('database_file', self._meta.config_defaults['database_file'])
timeout = kwargs.get('timeout', self._meta.config_defaults['timeout'])
self.connection = sqlite3.connect(db_file, timeout=timeout)
return self.connection
def execute(self, query, params=None):
"""Execute SQLite query."""
if not self.connection:
raise FrameworkError("Database not connected")
cursor = self.connection.cursor()
if params:
cursor.execute(query, params)
else:
cursor.execute(query)
return cursor.fetchall()
def close(self):
"""Close SQLite connection."""
if self.connection:
self.connection.close()
self.connection = Nonefrom cement import App, Controller, ex
class BaseController(Controller):
class Meta:
label = 'base'
@ex(help='test database operations')
def test_db(self):
"""Test database operations."""
# Get database handler through the app
db = self.app.handler.get('database', 'sqlite')()
try:
# Connect to database
db.connect(database_file='./test.db')
# Create table
db.execute('''
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
name TEXT,
email TEXT
)
''')
# Insert data
db.execute(
'INSERT INTO users (name, email) VALUES (?, ?)',
('John Doe', 'john@example.com')
)
# Query data
results = db.execute('SELECT * FROM users')
for row in results:
print(f'User: {row[1]}, Email: {row[2]}')
finally:
db.close()
class MyApp(App):
class Meta:
label = 'myapp'
base_controller = 'base'
handlers = [
BaseController,
SQLiteHandler
]
# Register the custom interface
def load(app):
app.interface.define(DatabaseInterface)
with MyApp() as app:
load(app)
app.run()from cement import Handler
import psycopg2
class PostgreSQLHandler(Handler, DatabaseInterface):
"""
PostgreSQL implementation of the database interface.
"""
class Meta:
label = 'postgresql'
interface = 'database'
config_defaults = {
'host': 'localhost',
'port': 5432,
'database': 'myapp',
'user': 'postgres',
'password': ''
}
def __init__(self, **kw):
super().__init__(**kw)
self.connection = None
def connect(self, **kwargs):
"""Connect to PostgreSQL database."""
config = self._meta.config_defaults.copy()
config.update(kwargs)
self.connection = psycopg2.connect(
host=config['host'],
port=config['port'],
database=config['database'],
user=config['user'],
password=config['password']
)
return self.connection
def execute(self, query, params=None):
"""Execute PostgreSQL query."""
if not self.connection:
raise FrameworkError("Database not connected")
cursor = self.connection.cursor()
cursor.execute(query, params)
return cursor.fetchall()
def close(self):
"""Close PostgreSQL connection."""
if self.connection:
self.connection.close()
self.connection = None
class MyApp(App):
class Meta:
label = 'myapp'
base_controller = 'base'
handlers = [
BaseController,
SQLiteHandler,
PostgreSQLHandler
]
def load(app):
app.interface.define(DatabaseInterface)
with MyApp() as app:
load(app)
# Can choose which database handler to use
sqlite_db = app.handler.get('database', 'sqlite')()
postgres_db = app.handler.get('database', 'postgresql')()
app.run()from cement import App, init_defaults
CONFIG = init_defaults('myapp')
CONFIG['myapp']['database_handler'] = 'sqlite' # Default handler
class MyApp(App):
class Meta:
label = 'myapp'
config_defaults = CONFIG
handler_override_options = {
'database': (['--database-handler'], {
'help': 'database handler to use',
'choices': ['sqlite', 'postgresql']
})
}
handlers = [
SQLiteHandler,
PostgreSQLHandler
]
def load(app):
app.interface.define(DatabaseInterface)
with MyApp() as app:
load(app)
# Handler can be overridden via command line:
# myapp --database-handler postgresql command
# Get the configured/overridden handler
handler_name = app.config.get('myapp', 'database_handler')
db = app.handler.get('database', handler_name)()
app.run()from cement import App
class MyApp(App):
class Meta:
label = 'myapp'
handlers = [SQLiteHandler, PostgreSQLHandler]
def load(app):
app.interface.define(DatabaseInterface)
with MyApp() as app:
load(app)
app.setup()
# Check if interface is defined
if app.interface.defined('database'):
print("Database interface is defined")
# Check if handlers are registered
if app.handler.registered('database', 'sqlite'):
print("SQLite handler is registered")
if app.handler.registered('database', 'postgresql'):
print("PostgreSQL handler is registered")
# List all handlers for interface
handlers = app.handler.list('database')
print(f"Available database handlers: {handlers}")Install with Tessl CLI
npx tessl i tessl/pypi-cement