tessl/pypi-nonebot2
An asynchronous Python bot framework for building cross-platform chatbots with plugin architecture and adapter support.
—
Pending
adapters-drivers.mddocs/
Adapters & Drivers
Base classes and interfaces for implementing protocol adapters and runtime drivers. These components provide the foundation for connecting NoneBot2 to different messaging platforms and runtime environments.
Capabilities
Adapter Base Class
Base class for implementing protocol adapters that connect NoneBot2 to messaging platforms.
class Adapter(abc.ABC):
"""Protocol adapter base class for connecting to messaging platforms."""
def __init__(self, driver: Driver, **kwargs: Any):
"""
Initialize adapter with driver instance.
Parameters:
- driver: Driver instance
- **kwargs: Additional adapter-specific configuration
"""
@abc.abstractmethod
def get_name(self) -> str:
"""
Get adapter name.
Returns:
str: Unique adapter name
"""
def bot_connect(self, bot: Bot) -> None:
"""
Handle bot connection event.
Parameters:
- bot: Bot instance that connected
"""
def bot_disconnect(self, bot: Bot) -> None:
"""
Handle bot disconnection event.
Parameters:
- bot: Bot instance that disconnected
"""
def setup_http_server(self, setup: HTTPServerSetup) -> None:
"""
Setup HTTP server for receiving webhooks.
Parameters:
- setup: HTTP server setup configuration
"""
def setup_websocket_server(self, setup: WebSocketServerSetup) -> None:
"""
Setup WebSocket server for receiving connections.
Parameters:
- setup: WebSocket server setup configuration
"""
async def request(self, setup: Request) -> Response:
"""
Make HTTP request.
Parameters:
- setup: Request configuration
Returns:
Response: HTTP response
"""
async def websocket(self, setup: Request) -> AsyncGenerator[WebSocket, None]:
"""
Create WebSocket connection.
Parameters:
- setup: WebSocket request configuration
Yields:
WebSocket: WebSocket connection
"""Bot Base Class
Base class for bot implementations that handle messages and API calls.
class Bot(abc.ABC):
"""Bot base class for handling messages and API calls."""
def __init__(self, adapter: Adapter, self_id: str):
"""
Initialize bot instance.
Parameters:
- adapter: Adapter instance that created this bot
- self_id: Bot's unique identifier
"""
async def call_api(self, api: str, **data: Any) -> Any:
"""
Call bot API method.
Parameters:
- api: API method name
- **data: API method parameters
Returns:
Any: API call result
Raises:
ApiNotAvailable: If API is not available
ActionFailed: If API call fails
NetworkError: If network error occurs
"""
def __getattr__(self, name: str) -> _ApiCall:
"""
Dynamic API call access.
Parameters:
- name: API method name
Returns:
_ApiCall: API call wrapper
"""
@property
def type(self) -> str:
"""
Get adapter name.
Returns:
str: Adapter name
"""
@property
def config(self) -> Config:
"""
Get global configuration.
Returns:
Config: Global configuration instance
"""Event Base Class
Base class for platform events with required abstract methods.
class Event(abc.ABC, BaseModel):
"""Event base class for platform events."""
@abc.abstractmethod
def get_type(self) -> str:
"""
Get event type.
Returns:
str: Event type string
"""
@abc.abstractmethod
def get_event_name(self) -> str:
"""
Get event name.
Returns:
str: Event name string
"""
@abc.abstractmethod
def get_event_description(self) -> str:
"""
Get event description.
Returns:
str: Human-readable event description
"""
@abc.abstractmethod
def get_user_id(self) -> str:
"""
Get user ID from event.
Returns:
str: User identifier
"""
@abc.abstractmethod
def get_session_id(self) -> str:
"""
Get session ID from event.
Returns:
str: Session identifier
"""
@abc.abstractmethod
def get_message(self) -> Message:
"""
Get message content from event.
Returns:
Message: Message object
"""
@abc.abstractmethod
def is_tome(self) -> bool:
"""
Check if event is directed to bot.
Returns:
bool: True if directed to bot
"""
def get_plaintext(self) -> str:
"""
Get plain text from message.
Returns:
str: Plain text content
"""
def get_log_string(self) -> str:
"""
Get log string representation.
Returns:
str: Log string
"""Driver Base Class
Base class for runtime drivers that manage bot lifecycle and connections.
class Driver(abc.ABC):
"""Base driver class for managing bot runtime."""
@property
@abc.abstractmethod
def type(self) -> str:
"""
Get driver type.
Returns:
str: Driver type string
"""
@property
@abc.abstractmethod
def logger(self) -> loguru.Logger:
"""
Get driver logger.
Returns:
loguru.Logger: Logger instance
"""
@abc.abstractmethod
def run(self, *args, **kwargs) -> None:
"""
Run the driver.
Parameters:
- *args: Runtime arguments
- **kwargs: Runtime keyword arguments
"""
def on_startup(self, func: _LIFESPAN_FUNC) -> _LIFESPAN_FUNC:
"""
Register startup handler.
Parameters:
- func: Startup handler function
Returns:
_LIFESPAN_FUNC: Handler function
"""
def on_shutdown(self, func: _LIFESPAN_FUNC) -> _LIFESPAN_FUNC:
"""
Register shutdown handler.
Parameters:
- func: Shutdown handler function
Returns:
_LIFESPAN_FUNC: Handler function
"""Driver Mixins
Mixin classes that add specific capabilities to drivers.
class Mixin(abc.ABC):
"""Base mixin class for extending driver capabilities."""
class ForwardMixin(Mixin):
"""Mixin for forward connection support (bot connects to platform)."""
class ReverseMixin(Mixin):
"""Mixin for reverse connection support (platform connects to bot)."""
class ASGIMixin(Mixin):
"""Mixin for ASGI server support."""
@property
@abc.abstractmethod
def server_app(self) -> Any:
"""
Get server application object.
Returns:
Any: Server app (FastAPI, Quart, etc.)
"""
@property
@abc.abstractmethod
def asgi(self) -> Any:
"""
Get ASGI application.
Returns:
Any: ASGI application
"""
class HTTPClientMixin(Mixin):
"""Mixin for HTTP client support."""
@abc.abstractmethod
async def request(self, setup: Request) -> Response:
"""
Make HTTP request.
Parameters:
- setup: Request configuration
Returns:
Response: HTTP response
"""
class WebSocketClientMixin(Mixin):
"""Mixin for WebSocket client support."""
@abc.abstractmethod
async def websocket(self, setup: Request) -> AsyncGenerator[WebSocket, None]:
"""
Create WebSocket connection.
Parameters:
- setup: WebSocket request configuration
Yields:
WebSocket: WebSocket connection
"""Composite Driver Classes
Pre-composed driver classes combining base driver with common mixins.
class ForwardDriver(Driver, ForwardMixin):
"""Driver with forward connection capability."""
class ReverseDriver(Driver, ReverseMixin):
"""Driver with reverse connection capability."""HTTP & WebSocket Models
Data models for HTTP and WebSocket communication.
class Request:
"""HTTP/WebSocket request model."""
method: str
url: URL
headers: dict[str, str]
cookies: Cookies
content: bytes
timeout: Optional[Timeout]
class Response:
"""HTTP response model."""
status_code: int
headers: dict[str, str]
content: bytes
def json(self) -> Any:
"""Parse response as JSON."""
def text(self) -> str:
"""Get response as text."""
class WebSocket:
"""WebSocket connection model."""
async def accept(self) -> None:
"""Accept WebSocket connection."""
async def close(self, code: int = 1000, reason: str = "") -> None:
"""Close WebSocket connection."""
async def receive(self) -> Union[str, bytes]:
"""Receive message from WebSocket."""
async def send(self, data: Union[str, bytes]) -> None:
"""Send message to WebSocket."""
class URL:
"""URL model for HTTP requests."""
scheme: str
host: str
port: Optional[int]
path: str
query: dict[str, str]
class Cookies:
"""Cookie management model."""
def get(self, key: str) -> Optional[str]:
"""Get cookie value."""
def set(self, key: str, value: str, **kwargs) -> None:
"""Set cookie value."""
class Timeout:
"""Request timeout configuration."""
connect: Optional[float]
read: Optional[float]
write: Optional[float]
total: Optional[float]Driver Utilities
Utility functions for working with drivers.
def combine_driver(driver: type[Driver], *mixins: type[Mixin]) -> type[Driver]:
"""
Combine driver class with mixin classes.
Parameters:
- driver: Base driver class
- *mixins: Mixin classes to combine
Returns:
type[Driver]: Combined driver class
"""Usage example:
from nonebot.drivers import Driver, ASGIMixin, HTTPClientMixin, combine_driver
# Create custom driver class
class MyDriver(Driver):
@property
def type(self):
return "my_driver"
@property
def logger(self):
return loguru.logger
def run(self, *args, **kwargs):
print("Running my driver")
# Combine with mixins
MyASGIDriver = combine_driver(MyDriver, ASGIMixin, HTTPClientMixin)
# Use combined driver
driver = MyASGIDriver()Types
Message Types
class Message(abc.ABC, List[MessageSegment]):
"""Message base class extending list of message segments."""
def __str__(self) -> str:
"""Get string representation."""
def __add__(self, other) -> "Message":
"""Concatenate messages."""
def extract_plain_text(self) -> str:
"""Extract plain text from message."""
class MessageSegment(abc.ABC, BaseModel):
"""Message segment base class for message components."""
type: str
"""Segment type."""
data: dict[str, Any]
"""Segment data."""
def __str__(self) -> str:
"""Get string representation."""
def is_text(self) -> bool:
"""Check if segment is text."""
class MessageTemplate(abc.ABC):
"""Message template base class for templated messages."""
def format(self, **kwargs) -> Message:
"""Format template with arguments."""Server Setup Types
class HTTPServerSetup:
"""HTTP server setup configuration."""
path: URL
method: str
name: str
class WebSocketServerSetup:
"""WebSocket server setup configuration."""
path: URL
name: str
class HTTPClientSession:
"""HTTP client session for making requests."""
async def request(self, setup: Request) -> Response:
"""Make HTTP request."""
async def websocket(self, setup: Request) -> AsyncGenerator[WebSocket, None]:
"""Create WebSocket connection."""Type Aliases
HTTPVersion = Literal["1.0", "1.1", "2.0"]
"""HTTP version type."""
_LIFESPAN_FUNC = Callable[[], Awaitable[None]]
"""Lifespan function type."""
_ApiCall = Callable[..., Awaitable[Any]]
"""API call wrapper type."""Install with Tessl CLI
npx tessl i tessl/pypi-nonebot2