API Reference

This page provides complete API documentation for d-back, automatically generated from the source code docstrings. All classes and methods include detailed descriptions, parameters, return values, and examples.

Introduction

The d-back API is organized into two main components:

• WebSocketServer: The main server class for handling WebSocket connections, HTTP requests, and message broadcasting
• MockDataProvider: Provides mock data and periodic background tasks for development and testing

All API documentation follows Google-style docstrings with comprehensive examples. For usage patterns and integration guides, see the User Guide.

WebSocketServer

The main server class for handling WebSocket connections, HTTP requests, and message broadcasting. This is your primary interface to d-back functionality.

WebSocketServer(port=3000, host='localhost')

WebSocket server for managing real-time connections and broadcasting messages.

This server handles WebSocket connections for Discord-like applications, providing real-time communication, user presence management, message broadcasting, and static file serving. It supports custom data providers through callback registration and includes mock data functionality for testing.

The server automatically detects websockets library version and enables HTTP support when available (websockets 10.0+ with Python 3.8+).

Attributes:

Name	Type	Description
port	int	The port number the server listens on.
host	str	The hostname or IP address the server binds to.
server		The websockets server instance.
connections	set	Set of active WebSocket connections.
static_dir	Path	Directory path for serving static files.
mock_provider	MockDataProvider	Provider for mock test data.

Example

Basic usage::

server = WebSocketServer(port=3000, host="localhost")
await server.run_forever()

With custom data callbacks::

def get_servers():
    return {
        "server1": {"id": "server1", "name": "My Server", "default": True}
    }

def get_users(server_id):
    return {
        "user123": {
            "uid": "user123",
            "username": "JohnDoe",
            "status": "online",
            "roleColor": "#3498db"
        }
    }

server = WebSocketServer()
server.on_get_server_data(get_servers)
server.on_get_user_data(get_users)
await server.run_forever()

Initialize the WebSocket server.

Parameters:

Name	Type	Description	Default
port	int	The port number to listen on. Defaults to 3000.	3000
host	str	The hostname or IP address to bind to. Defaults to "localhost". Use "0.0.0.0" to accept connections from any interface.	'localhost'

Note

The server initializes with mock data provider by default. Register custom callbacks using on_get_server_data() and on_get_user_data() to override mock data behavior.

Source code in d_back/server.py

def __init__(self, port: int = 3000, host: str = "localhost"):
    """Initialize the WebSocket server.

    Args:
        port: The port number to listen on. Defaults to 3000.
        host: The hostname or IP address to bind to. Defaults to "localhost".
              Use "0.0.0.0" to accept connections from any interface.

    Note:
        The server initializes with mock data provider by default. Register custom
        callbacks using on_get_server_data() and on_get_user_data() to override
        mock data behavior.
    """
    self.port = port
    self.host = host
    self.server = None  # WebSocket server instance
    self.connections: set = set()  # Store active connections
    self._on_get_server_data: Optional[Callable[[], Awaitable[Dict[str, Dict[str, Any]]]]] = None
    self._on_get_user_data: Optional[Callable[[str], Awaitable[Dict[str, Dict[str, Any]]]]] = None
    self._on_static_request: Optional[Callable[[str], Awaitable[Optional[Tuple[str, str]]]]] = None
    self.static_dir = Path(__file__).parent / "dist"  # Default static directory
    self._on_validate_discord_user: Optional[Callable[[str, Dict[str, Any], str], Awaitable[bool]]] = None
    self._on_get_client_id: Optional[Callable[[str], Awaitable[str]]] = None
    self.mock_provider = MockDataProvider(self)  # Mock data provider

start() async

Start the WebSocket server and wait for it to close.

This method initializes the websockets server with the configured host and port, registers the connection handler and HTTP request processor, and then waits for the server to close.

Returns:

Type	Description
None	None

Raises:

Type	Description
OSError	If the port is already in use or network binding fails.
Exception	For other server startup failures.

Note

This method blocks until the server is stopped. Use run_forever() for a more convenient async context manager approach.

Source code in d_back/server.py

async def start(self) -> None:
    """Start the WebSocket server and wait for it to close.

    This method initializes the websockets server with the configured host and port,
    registers the connection handler and HTTP request processor, and then waits
    for the server to close.

    Returns:
        None

    Raises:
        OSError: If the port is already in use or network binding fails.
        Exception: For other server startup failures.

    Note:
        This method blocks until the server is stopped. Use run_forever() for
        a more convenient async context manager approach.
    """
    self.server = await websockets.serve(
        self._handler, 
        self.host, 
        self.port, 
        process_request=self._process_request
    )
    print(f"WebSocket server started on ws://{self.host}:{self.port}")
    await self.server.wait_closed()

stop() async

Gracefully stop the WebSocket server.

This method closes the server and waits for all existing connections to terminate. Active connections are allowed to finish their current operations before shutdown.

Returns:

Type	Description
None	None

Note

After calling this method, the server can be restarted by calling start() again.

Source code in d_back/server.py

async def stop(self) -> None:
    """Gracefully stop the WebSocket server.

    This method closes the server and waits for all existing connections to terminate.
    Active connections are allowed to finish their current operations before shutdown.

    Returns:
        None

    Note:
        After calling this method, the server can be restarted by calling start() again.
    """
    if self.server:
        self.server.close()
        await self.server.wait_closed()
        print("WebSocket server stopped")

broadcast_message(server, uid, message, channel) async

Broadcast a chat message to all clients connected to the specified server.

Sends a message from a user to all WebSocket clients currently connected to the same Discord server. Automatically handles connection failures and removes closed connections.

Parameters:

Name	Type	Description	Default
server	str	The Discord server ID where the message is sent.	required
uid	str	The user ID of the message sender.	required
message	str	The text content of the message.	required
channel	str	The channel ID where the message is posted.	required

Returns:

Type	Description
None	None

Raises:

Type	Description
ConnectionClosed	When attempting to send to a closed connection (handled internally, connection is removed).

Examples:

await server.broadcast_message(
    server="232769614004748288",
    uid="user123",
    message="Hello everyone!",
    channel="general"
)

Source code in d_back/server.py

async def broadcast_message(self, server: str, uid: str, message: str, channel: str) -> None:
    """Broadcast a chat message to all clients connected to the specified server.

    Sends a message from a user to all WebSocket clients currently connected to
    the same Discord server. Automatically handles connection failures and removes
    closed connections.

    Args:
        server: The Discord server ID where the message is sent.
        uid: The user ID of the message sender.
        message: The text content of the message.
        channel: The channel ID where the message is posted.

    Returns:
        None

    Raises:
        websockets.ConnectionClosed: When attempting to send to a closed connection
            (handled internally, connection is removed).

    Examples:

        await server.broadcast_message(
            server="232769614004748288",
            uid="user123",
            message="Hello everyone!",
            channel="general"
        )
    """
    # Filter connections to only include those connected to the specified server
    # Use discordServer like the original implementation
    server_connections = [ws for ws in self.connections if hasattr(ws, 'discordServer') and ws.discordServer == server]

    if not server_connections:
        print(f"[INFO] No connections to broadcast to for server: {server}")
        return

    msg = {
        "type": "message",
        "server": server,
        "data": {
            "uid": uid,
            "message": message,
            "channel": channel
        }
    }

    print(f"[BROADCAST] Sending message to {len(server_connections)} connections on server {server}: {message}")

    # Create a copy to avoid modification during iteration
    connections_copy = server_connections.copy()

    for websocket in connections_copy:
        try:
            await websocket.send(json.dumps(msg))
        except websockets.ConnectionClosed:
            print("[INFO] Removed closed connection during broadcast")
            # Remove closed connections
            self.connections.discard(websocket)
        except Exception as e:
            print(f"[ERROR] Failed to send message to connection: {e}")
            # Optionally remove problematic connections
            self.connections.discard(websocket)

on_get_server_data(callback)

Register a callback to provide server configuration data.

The callback function will be called to retrieve the list of available Discord servers and their configurations. This overrides the default mock data provider.

Parameters:

Name	Type	Description	Default
callback	Callable[[], Awaitable[Dict[str, Dict[str, Any]]]]	An async callable that returns a dictionary of server configurations. Expected signature: async def callback() -> Dict[str, Dict[str, Any]] Each server dict should contain: id, name, default (bool), passworded (bool)	required

Returns:

Type	Description
None	None

Examples:

async def get_servers():
    return {
        "server1": {
            "id": "server1",
            "name": "My Server",
            "default": True,
            "passworded": False
        }
    }

server.on_get_server_data(get_servers)

Source code in d_back/server.py

def on_get_server_data(self, callback: Callable[[], Awaitable[Dict[str, Dict[str, Any]]]]) -> None:
    """Register a callback to provide server configuration data.

    The callback function will be called to retrieve the list of available Discord
    servers and their configurations. This overrides the default mock data provider.

    Args:
        callback: An async callable that returns a dictionary of server configurations.
            Expected signature: async def callback() -> Dict[str, Dict[str, Any]]
            Each server dict should contain: id, name, default (bool), passworded (bool)

    Returns:
        None

    Examples:

        async def get_servers():
            return {
                "server1": {
                    "id": "server1",
                    "name": "My Server",
                    "default": True,
                    "passworded": False
                }
            }

        server.on_get_server_data(get_servers)
    """
    self._on_get_server_data = callback

on_get_user_data(callback)

Register a callback to provide user data for a specific server.

The callback function will be called to retrieve user information for clients connecting to a Discord server. This overrides the default mock data provider.

Parameters:

Name	Type	Description	Default
callback	Callable[[str], Awaitable[Dict[str, Dict[str, Any]]]]	An async callable that takes a server ID and returns user data. Expected signature: async def callback(server_id: str) -> Dict[str, Dict[str, Any]] Each user dict should contain: uid, username, status, roleColor	required

Returns:

Type	Description
None	None

Examples:

async def get_users(server_id):
    return {
        "user123": {
            "uid": "user123",
            "username": "JohnDoe",
            "status": "online",
            "roleColor": "#3498db"
        }
    }

server.on_get_user_data(get_users)

Source code in d_back/server.py

def on_get_user_data(self, callback: Callable[[str], Awaitable[Dict[str, Dict[str, Any]]]]) -> None:
    """Register a callback to provide user data for a specific server.

    The callback function will be called to retrieve user information for clients
    connecting to a Discord server. This overrides the default mock data provider.

    Args:
        callback: An async callable that takes a server ID and returns user data.
            Expected signature: async def callback(server_id: str) -> Dict[str, Dict[str, Any]]
            Each user dict should contain: uid, username, status, roleColor

    Returns:
        None

    Examples:

        async def get_users(server_id):
            return {
                "user123": {
                    "uid": "user123",
                    "username": "JohnDoe",
                    "status": "online",
                    "roleColor": "#3498db"
                }
            }

        server.on_get_user_data(get_users)
    """
    self._on_get_user_data = callback

on_static_request(callback)

Register a callback for custom static file handling.

The callback function allows you to serve custom content for specific paths, overriding the default static file handler. Return None to use default handling.

Parameters:

Name	Type	Description	Default
callback	Callable[[str], Awaitable[Optional[Tuple[str, str]]]]	An async callable that takes a path and returns custom content. Expected signature: async def callback(path: str) -> Optional[Tuple[str, str]] Return None to let default handler process the request, or return (content_type, content) to serve custom content.	required

Returns:

Type	Description
None	None

Examples:

async def custom_handler(path):
    if path == "/api/custom":
        return "application/json", '{"status": "ok"}'
    return None  # Let default handler take over

server.on_static_request(custom_handler)

Source code in d_back/server.py

def on_static_request(self, callback: Callable[[str], Awaitable[Optional[Tuple[str, str]]]]) -> None:
    """Register a callback for custom static file handling.

    The callback function allows you to serve custom content for specific paths,
    overriding the default static file handler. Return None to use default handling.

    Args:
        callback: An async callable that takes a path and returns custom content.
            Expected signature: async def callback(path: str) -> Optional[Tuple[str, str]]
            Return None to let default handler process the request, or
            return (content_type, content) to serve custom content.

    Returns:
        None

    Examples:

        async def custom_handler(path):
            if path == "/api/custom":
                return "application/json", '{"status": "ok"}'
            return None  # Let default handler take over

        server.on_static_request(custom_handler)
    """
    self._on_static_request = callback

on_validate_discord_user(callback)

Register a callback to validate Discord OAuth users.

The callback function will be called to verify Discord OAuth tokens and validate user permissions for accessing the WebSocket server.

Parameters:

Name	Type	Description	Default
callback	Callable[[str, Dict[str, Any], str], Awaitable[bool]]	An async callable that validates a Discord OAuth token and user info. Expected signature: async def callback(token: str, user_info: Dict[str, Any], server_id: str) -> bool Should return True if the user is valid and authorized.	required

Returns:

Type	Description
None	None

Examples:

async def validate_user(token, user_info, server_id):
    # Verify token with Discord API
    # Check user permissions for the specific server
    return is_valid and is_authorized

server.on_validate_discord_user(validate_user)

Source code in d_back/server.py

def on_validate_discord_user(self, callback: Callable[[str, Dict[str, Any], str], Awaitable[bool]]) -> None:
    """Register a callback to validate Discord OAuth users.

    The callback function will be called to verify Discord OAuth tokens and
    validate user permissions for accessing the WebSocket server.

    Args:
        callback: An async callable that validates a Discord OAuth token and user info.
            Expected signature: async def callback(token: str, user_info: Dict[str, Any], server_id: str) -> bool
            Should return True if the user is valid and authorized.

    Returns:
        None

    Examples:

        async def validate_user(token, user_info, server_id):
            # Verify token with Discord API
            # Check user permissions for the specific server
            return is_valid and is_authorized

        server.on_validate_discord_user(validate_user)
    """
    self._on_validate_discord_user = callback

on_get_client_id(callback)

Register a callback to provide the OAuth2 client ID.

The callback function should return the Discord OAuth2 application client ID used for authentication. This is sent to connecting clients for OAuth flow.

Parameters:

Name	Type	Description	Default
callback	Callable[[str], Awaitable[str]]	An async callable that returns the OAuth2 client ID for a server. Expected signature: async def callback(server_id: str) -> str Should return the Discord application client ID string.	required

Returns:

Type	Description
None	None

Examples:

async def get_client_id(server_id):
    return "123456789012345678"

server.on_get_client_id(get_client_id)

Source code in d_back/server.py

def on_get_client_id(self, callback: Callable[[str], Awaitable[str]]) -> None:
    """Register a callback to provide the OAuth2 client ID.

    The callback function should return the Discord OAuth2 application client ID
    used for authentication. This is sent to connecting clients for OAuth flow.

    Args:
        callback: An async callable that returns the OAuth2 client ID for a server.
            Expected signature: async def callback(server_id: str) -> str
            Should return the Discord application client ID string.

    Returns:
        None

    Examples:

        async def get_client_id(server_id):
            return "123456789012345678"

        server.on_get_client_id(get_client_id)
    """
    self._on_get_client_id = callback

run_forever() async

Run the server forever with automatic HTTP support detection.

This method starts the WebSocket server and automatically detects whether HTTP static file serving is supported based on the Python version and websockets library version. Falls back to WebSocket-only mode if HTTP support is unavailable.

HTTP support requires

• Python 3.8 or higher
• websockets 10.0 or higher
• Available HTTP import modules (websockets.http11 or websockets.http)

Returns:

Type	Description
None	None

Raises:

Type	Description
Exception	If server fails to start on the configured host and port.
KeyboardInterrupt	When server is interrupted by user (handled gracefully).

Examples:

server = WebSocketServer(port=3000)
await server.run_forever()  # Runs until interrupted

Note

This method runs indefinitely until interrupted. Use asyncio.create_task() if you need to run other async operations concurrently.

Source code in d_back/server.py

async def run_forever(self) -> None:
    """Run the server forever with automatic HTTP support detection.

    This method starts the WebSocket server and automatically detects whether
    HTTP static file serving is supported based on the Python version and websockets
    library version. Falls back to WebSocket-only mode if HTTP support is unavailable.

    HTTP support requires:
        - Python 3.8 or higher
        - websockets 10.0 or higher
        - Available HTTP import modules (websockets.http11 or websockets.http)

    Returns:
        None

    Raises:
        Exception: If server fails to start on the configured host and port.
        KeyboardInterrupt: When server is interrupted by user (handled gracefully).

    Examples:

        server = WebSocketServer(port=3000)
        await server.run_forever()  # Runs until interrupted

    Note:
        This method runs indefinitely until interrupted. Use asyncio.create_task()
        if you need to run other async operations concurrently.
    """
    # For Python 3.8+ compatibility, start with WebSocket-only mode
    # and only try HTTP if we're confident it will work
    has_http_support = False

    try:
        import websockets
        # Check websockets version
        websockets_version = tuple(map(int, websockets.__version__.split('.')[:2]))

        # Try HTTP support on Python 3.8+ with websockets 10.0+
        import sys
        python_version = sys.version_info[:2]

        if python_version >= (3, 8) and websockets_version >= (10, 0):
            try:
                # Quick test of HTTP imports
                from websockets.http11 import Response  # noqa: F401
                from websockets.http import Headers  # noqa: F401
                has_http_support = True
                print(f"[DEBUG] HTTP support enabled (Python {python_version}, websockets {websockets.__version__})")
            except ImportError:
                try:
                    from websockets.http import Response, Headers  # noqa: F401
                    has_http_support = True
                    print("[DEBUG] HTTP support enabled with fallback imports")
                except ImportError:
                    print("[DEBUG] HTTP imports not available, using WebSocket-only mode")
        else:
            print(f"[DEBUG] WebSocket-only mode (Python {python_version}, websockets {websockets.__version__} - version too old for HTTP)")

    except Exception as e:
        print(f"[WARNING] Error checking HTTP support, falling back to WebSocket-only: {e}")
        has_http_support = False

    if has_http_support:
        try:
            async with websockets.serve(
                self._handler, 
                self.host, 
                self.port, 
                process_request=self._process_request
            ):
                print(f"Mock WebSocket server running on ws://{self.host}:{self.port} (with HTTP support)")
                await asyncio.Future()  # run forever
        except Exception as e:
            print(f"[WARNING] Failed to start with HTTP support: {e}")
            print("[INFO] Falling back to WebSocket-only mode")
            has_http_support = False

    if not has_http_support:
        async with websockets.serve(
            self._handler, 
            self.host, 
            self.port
        ):
            print(f"Mock WebSocket server running on ws://{self.host}:{self.port} (WebSocket-only mode)")
            await asyncio.Future()  # run forever

broadcast_presence(server, uid, status, username=None, role_color=None, delete=False) async

Broadcast a user presence update to all clients connected to a server.

Sends presence information (online status, username, role color) to all WebSocket connections associated with the specified Discord server. Used to notify clients when users come online, go offline, or change status.

Parameters:

Name	Type	Description	Default
server	str	Discord server ID to broadcast to	required
uid	str	User ID whose presence is being updated	required
status	str	User's current status ("online", "idle", "dnd", "offline")	required
username	str	Optional username to include in the update	None
role_color	str	Optional hex color code for the user's role (e.g., "#ff6b6b")	None
delete	bool	If True, indicates the user should be removed from the presence list	False

Returns:

Type	Description
None	None

Examples:

# Broadcast user coming online
await server.broadcast_presence(
    server="232769614004748288",
    uid="123456789012345001",
    status="online",
    username="vegeta897",
    role_color="#ff6b6b"
)

# Broadcast user going offline
await server.broadcast_presence(
    server="232769614004748288",
    uid="123456789012345001",
    status="offline",
    delete=True
)

Source code in d_back/server.py

async def broadcast_presence(self, server: str, uid: str, status: str, username: str = None, role_color: str = None, delete: bool = False) -> None:
    """Broadcast a user presence update to all clients connected to a server.

    Sends presence information (online status, username, role color) to all
    WebSocket connections associated with the specified Discord server. Used
    to notify clients when users come online, go offline, or change status.

    Args:
        server: Discord server ID to broadcast to
        uid: User ID whose presence is being updated
        status: User's current status ("online", "idle", "dnd", "offline")
        username: Optional username to include in the update
        role_color: Optional hex color code for the user's role (e.g., "#ff6b6b")
        delete: If True, indicates the user should be removed from the presence list

    Returns:
        None

    Examples:

        # Broadcast user coming online
        await server.broadcast_presence(
            server="232769614004748288",
            uid="123456789012345001",
            status="online",
            username="vegeta897",
            role_color="#ff6b6b"
        )

        # Broadcast user going offline
        await server.broadcast_presence(
            server="232769614004748288",
            uid="123456789012345001",
            status="offline",
            delete=True
        )
    """
    # Filter connections to only include those connected to the specified server
    server_connections = [ws for ws in self.connections if hasattr(ws, 'discordServer') and ws.discordServer == server]

    if not server_connections:
        print(f"[INFO] No connections to broadcast presence to for server: {server}")
        return

    presence_data = {
        "uid": uid,
        "status": status
    }

    if username:
        presence_data["username"] = username
    if role_color:
        presence_data["roleColor"] = role_color
    if delete:
        presence_data["delete"] = True

    msg = {
        "type": "presence",
        "server": server,
        "data": presence_data
    }

    print(f"[BROADCAST] Sending presence update to {len(server_connections)} connections on server {server}: {uid} -> {status}")

    # Create a copy to avoid modification during iteration
    connections_copy = server_connections.copy()

    for websocket in connections_copy:
        try:
            await websocket.send(json.dumps(msg))
        except websockets.ConnectionClosed:
            print("[INFO] Removed closed connection during presence broadcast")
            # Remove closed connections
            self.connections.discard(websocket)
        except Exception as e:
            print(f"[ERROR] Failed to send presence update to connection: {e}")
            # Optionally remove problematic connections
            self.connections.discard(websocket)

broadcast_client_id_update(server, client_id) async

Broadcast an OAuth2 client ID update to all clients connected to a server.

Sends the Discord OAuth2 application client ID to all WebSocket connections associated with the specified Discord server. Used to dynamically update the client ID for OAuth authentication flows.

Parameters:

Name	Type	Description	Default
server	str	Discord server ID to broadcast to	required
client_id	str	Discord OAuth2 application client ID to send	required

Returns:

Type	Description
None	None

Examples:

# Update client ID for all connections to a server
await server.broadcast_client_id_update(
    server="232769614004748288",
    client_id="123456789012345678"
)

Source code in d_back/server.py

async def broadcast_client_id_update(self, server: str, client_id: str) -> None:
    """Broadcast an OAuth2 client ID update to all clients connected to a server.

    Sends the Discord OAuth2 application client ID to all WebSocket connections
    associated with the specified Discord server. Used to dynamically update
    the client ID for OAuth authentication flows.

    Args:
        server: Discord server ID to broadcast to
        client_id: Discord OAuth2 application client ID to send

    Returns:
        None

    Examples:

        # Update client ID for all connections to a server
        await server.broadcast_client_id_update(
            server="232769614004748288",
            client_id="123456789012345678"
        )
    """
    # Filter connections to only include those connected to the specified server
    server_connections = [ws for ws in self.connections if hasattr(ws, 'discordServer') and ws.discordServer == server]

    if not server_connections:
        print(f"[INFO] No connections to broadcast client ID update to for server: {server}")
        return

    msg = {
        "type": "update-clientid",
        "server": server,
        "data": {
            "clientId": client_id
        }
    }

    print(f"[BROADCAST] Sending client ID update to {len(server_connections)} connections on server {server}: {client_id}")

    # Create a copy to avoid modification during iteration
    connections_copy = server_connections.copy()

    for websocket in connections_copy:
        try:
            await websocket.send(json.dumps(msg))
        except websockets.ConnectionClosed:
            print("[INFO] Removed closed connection during client ID broadcast")
            # Remove closed connections
            self.connections.discard(websocket)
        except Exception as e:
            print(f"[ERROR] Failed to send client ID update to connection: {e}")
            # Optionally remove problematic connections
            self.connections.discard(websocket)

MockDataProvider

Provides mock data and periodic background tasks for development and testing. This class is used automatically when custom callbacks are not registered.

MockDataProvider(server_instance)

Provides mock data and periodic background tasks for WebSocket server testing.

This class generates test data that simulates a Discord-like server environment with multiple servers, users, and real-time updates. It provides both static data (user lists, server configurations) and dynamic behaviors (status updates, message broadcasts) for development and testing purposes.

The mock provider includes predefined data for several test servers

• D-World: Main test server with 4 users
• Docs (WIP): Documentation server with 1 user
• OAuth2 Protected Server: Authentication testing server
• My Repos: Repository showcase server with 21 users

Attributes:

Name	Type	Description
server		Reference to the WebSocketServer instance using this provider.

Examples:

Basic usage:

server = WebSocketServer()
mock_provider = MockDataProvider(server)

# Get user data for a specific server
users = mock_provider.get_mock_user_data("232769614004748288")
# Returns: {"uid1": {"uid": "...", "username": "...", ...}, ...}

# Get all available servers
servers = mock_provider.get_mock_server_data()
# Returns: {"server_id": {"id": "...", "name": "...", ...}, ...}

Note

The mock data provider is automatically instantiated by WebSocketServer. You typically don't need to create instances manually unless testing the provider in isolation.

Initialize the mock data provider.

Parameters:

Name	Type	Description	Default
server_instance	WebSocketServer	The WebSocketServer instance that owns this provider. Used for accessing server methods like _random_status().	required

Source code in d_back/mock/data.py

def __init__(self, server_instance: 'WebSocketServer'):
    """Initialize the mock data provider.

    Args:
        server_instance: The WebSocketServer instance that owns this provider.
            Used for accessing server methods like _random_status().
    """
    self.server = server_instance

get_mock_user_data(discord_server_id=None)

Get mock user data for a specific Discord server.

Returns a dictionary of mock users with their profile information including user ID, username, online status, and role color. Each server has a predefined set of users for consistent testing.

Parameters:

Name	Type	Description	Default
discord_server_id	str	Optional Discord server ID to get users for. Supported IDs: - "232769614004748288": D-World server (4 users) - "482241773318701056": Docs (WIP) server (1 user) - "123456789012345678": OAuth2 Protected server (1 user) - "987654321098765432": My Repos server (21 users) If None or unknown, returns empty dict.	None

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: Dictionary mapping user IDs to user data objects. Each user object contains: - uid (str): Unique user identifier - username (str): Display name - status (str): Online status ("online", "idle", "dnd", "offline") - roleColor (str): Hex color code for user's role (e.g., "#ff6b6b")

Examples:

provider = MockDataProvider(server)
users = provider.get_mock_user_data("232769614004748288")
# Returns:
# {
#     "123456789012345001": {
#         "uid": "123456789012345001",
#         "username": "vegeta897",
#         "status": "online",
#         "roleColor": "#ff6b6b"
#     },
#     ...
# }

Source code in d_back/mock/data.py

def get_mock_user_data(self, discord_server_id: str = None) -> Dict[str, Any]:
    """Get mock user data for a specific Discord server.

    Returns a dictionary of mock users with their profile information including
    user ID, username, online status, and role color. Each server has a predefined
    set of users for consistent testing.

    Args:
        discord_server_id: Optional Discord server ID to get users for.
            Supported IDs:
            - "232769614004748288": D-World server (4 users)
            - "482241773318701056": Docs (WIP) server (1 user)
            - "123456789012345678": OAuth2 Protected server (1 user)
            - "987654321098765432": My Repos server (21 users)
            If None or unknown, returns empty dict.

    Returns:
        Dict[str, Any]: Dictionary mapping user IDs to user data objects.
            Each user object contains:
            - uid (str): Unique user identifier
            - username (str): Display name
            - status (str): Online status ("online", "idle", "dnd", "offline")
            - roleColor (str): Hex color code for user's role (e.g., "#ff6b6b")

    Examples:

        provider = MockDataProvider(server)
        users = provider.get_mock_user_data("232769614004748288")
        # Returns:
        # {
        #     "123456789012345001": {
        #         "uid": "123456789012345001",
        #         "username": "vegeta897",
        #         "status": "online",
        #         "roleColor": "#ff6b6b"
        #     },
        #     ...
        # }
    """

    # D-World server users (default)
    if discord_server_id == "232769614004748288":
        return {
            "123456789012345001": {
                "uid": "123456789012345001",
                "username": "vegeta897",
                "status": "online",
                "roleColor": "#ff6b6b"
            },
            "123456789012345002": {
                "uid": "123456789012345002",
                "username": "Cog-Creators",
                "status": "idle",
                "roleColor": "#4ecdc4"
            },
            "123456789012345003": {
                "uid": "123456789012345003",
                "username": "d-zone-org",
                "status": "dnd",
                "roleColor": "#45b7d1"
            },
            "123456789012345004": {
                "uid": "123456789012345004",
                "username": "NNTin",
                "status": "online",
                "roleColor": "#96ceb4"
            }
        }

    # Docs (WIP) server users
    elif discord_server_id == "482241773318701056":
        return {
            "223456789012345001": {
                "uid": "223456789012345001",
                "username": "nntin.xyz/me",
                "status": "online",
                "roleColor": "#feca57"
            }
        }

    # OAuth2 Protected server users
    elif discord_server_id == "123456789012345678":
        return {
            "323456789012345001": {
                "uid": "323456789012345001",
                "username": "NNTin",
                "status": "online",
                "roleColor": "#ff9ff3"
            }
        }

    # My Repos server users
    elif discord_server_id == "987654321098765432":
        return {
            "423456789012345001": {
                "uid": "423456789012345001",
                "username": "me",
                "status": "online",
                "roleColor": "#54a0ff"
            },
            "423456789012345002": {
                "uid": "423456789012345002",
                "username": "nntin.github.io",
                "status": "idle",
                "roleColor": "#5f27cd"
            },
            "423456789012345003": {
                "uid": "423456789012345003",
                "username": "d-zone",
                "status": "online",
                "roleColor": "#00d2d3"
            },
            "423456789012345004": {
                "uid": "423456789012345004",
                "username": "d-back",
                "status": "dnd",
                "roleColor": "#ff6348"
            },
            "423456789012345005": {
                "uid": "423456789012345005",
                "username": "d-cogs",
                "status": "online",
                "roleColor": "#ff4757"
            },
            "423456789012345006": {
                "uid": "423456789012345006",
                "username": "Cubify-Reddit",
                "status": "offline",
                "roleColor": "#3742fa"
            },
            "423456789012345007": {
                "uid": "423456789012345007",
                "username": "Dota-2-Emoticons",
                "status": "idle",
                "roleColor": "#2ed573"
            },
            "423456789012345008": {
                "uid": "423456789012345008",
                "username": "Dota-2-Reddit-Flair-Mosaic",
                "status": "online",
                "roleColor": "#ffa502"
            },
            "423456789012345009": {
                "uid": "423456789012345009",
                "username": "Red-kun",
                "status": "dnd",
                "roleColor": "#ff3838"
            },
            "423456789012345010": {
                "uid": "423456789012345010",
                "username": "Reply-Dota-2-Reddit",
                "status": "online",
                "roleColor": "#ff9f43"
            },
            "423456789012345011": {
                "uid": "423456789012345011",
                "username": "Reply-LoL-Reddit",
                "status": "idle",
                "roleColor": "#0abde3"
            },
            "423456789012345012": {
                "uid": "423456789012345012",
                "username": "crosku",
                "status": "online",
                "roleColor": "#006ba6"
            },
            "423456789012345013": {
                "uid": "423456789012345013",
                "username": "dev-tracker-reddit",
                "status": "offline",
                "roleColor": "#8e44ad"
            },
            "423456789012345014": {
                "uid": "423456789012345014",
                "username": "discord-logo",
                "status": "online",
                "roleColor": "#7289da"
            },
            "423456789012345015": {
                "uid": "423456789012345015",
                "username": "discord-twitter-bot",
                "status": "idle",
                "roleColor": "#1da1f2"
            },
            "423456789012345016": {
                "uid": "423456789012345016",
                "username": "discord-web-bridge",
                "status": "dnd",
                "roleColor": "#2c2f33"
            },
            "423456789012345017": {
                "uid": "423456789012345017",
                "username": "pasteindex",
                "status": "online",
                "roleColor": "#f39c12"
            },
            "423456789012345018": {
                "uid": "423456789012345018",
                "username": "pasteview",
                "status": "idle",
                "roleColor": "#e74c3c"
            },
            "423456789012345019": {
                "uid": "423456789012345019",
                "username": "shell-kun",
                "status": "online",
                "roleColor": "#1abc9c"
            },
            "423456789012345020": {
                "uid": "423456789012345020",
                "username": "tracker-reddit-discord",
                "status": "offline",
                "roleColor": "#9b59b6"
            },
            "423456789012345021": {
                "uid": "423456789012345021",
                "username": "twitter-backend",
                "status": "online",
                "roleColor": "#1da1f2"
            }
        }

    # Fallback: return empty if unknown server
    return {}

get_mock_server_data()

Get mock server data with all available Discord servers.

Returns a dictionary mapping Discord server IDs to server configuration objects. Used for testing server selection, display, and navigation features.

Returns:

Type	Description
Dict[str, Any]	Dict[str, Dict[str, Any]]: Dictionary mapping Discord server snowflake IDs to server configuration objects. Each server object contains: - id (str): Internal server identifier - name (str): Server display name - passworded (bool): Whether OAuth2 authentication is required - default (bool, optional): Whether this is the default server

Examples:

provider = MockDataProvider(server)
servers = provider.get_mock_server_data()
# Returns:
# {
#     "232769614004748288": {
#         "id": "dworld",
#         "name": "D-World",
#         "default": True,
#         "passworded": False
#     },
#     "482241773318701056": {
#         "id": "docs",
#         "name": "Docs (WIP)",
#         "passworded": False
#     },
#     ...
# }

for snowflake_id, server_info in servers.items():
    print(f"{server_info['name']} (ID: {snowflake_id})")

Source code in d_back/mock/data.py

def get_mock_server_data(self) -> Dict[str, Any]:
    """Get mock server data with all available Discord servers.

    Returns a dictionary mapping Discord server IDs to server configuration
    objects. Used for testing server selection, display, and navigation features.

    Returns:
        Dict[str, Dict[str, Any]]: Dictionary mapping Discord server snowflake IDs
            to server configuration objects. Each server object contains:
            - id (str): Internal server identifier
            - name (str): Server display name
            - passworded (bool): Whether OAuth2 authentication is required
            - default (bool, optional): Whether this is the default server

    Examples:

        provider = MockDataProvider(server)
        servers = provider.get_mock_server_data()
        # Returns:
        # {
        #     "232769614004748288": {
        #         "id": "dworld",
        #         "name": "D-World",
        #         "default": True,
        #         "passworded": False
        #     },
        #     "482241773318701056": {
        #         "id": "docs",
        #         "name": "Docs (WIP)",
        #         "passworded": False
        #     },
        #     ...
        # }

        for snowflake_id, server_info in servers.items():
            print(f"{server_info['name']} (ID: {snowflake_id})")
    """
    return {
        "232769614004748288": {
            "id": "dworld",
            "name": "D-World",
            "default": True,
            "passworded": False
        },
        "482241773318701056": {
            "id": "docs", 
            "name": "Docs (WIP)",
            "passworded": False
        },
        "123456789012345678": {
            "id": "oauth",
            "name": "OAuth2 Protected Server",
            "passworded": True
        },
        "987654321098765432": {
            "id": "repos",
            "name": "My Repos",
            "passworded": False
        }
    }

periodic_status_updates(websocket) async

Periodically send mock user status changes to a connected client.

Background task that continuously generates random user status updates and sends them to the specified WebSocket connection. Simulates realistic user activity by randomly changing user statuses every 4 seconds.

This method runs until the WebSocket connection is closed or the task is cancelled. It selects random users from the current server and updates their online status (online, idle, dnd, offline).

Parameters:

Name	Type	Description	Default
websocket	WebSocketServerProtocol	WebSocket connection to send updates to. Must have a discordServer attribute identifying which server's users to update.	required

Note

This is a coroutine that runs indefinitely for the lifetime of the WebSocket connection. It automatically handles ConnectionClosed exceptions gracefully.

Examples:

# Called automatically by the server for each connection
task = asyncio.create_task(
    mock_provider.periodic_status_updates(websocket)
)

# Updates will be sent in this format:
# {
#     "type": "presence",
#     "server": "232769614004748288",
#     "data": {
#         "uid": "123456789012345001",
#         "status": "online"
#     }
# }

Raises:

Type	Description
ConnectionClosed	When the WebSocket connection is closed. This exception is caught and handled gracefully.

Source code in d_back/mock/data.py

async def periodic_status_updates(self, websocket: 'WebSocketServerProtocol') -> None:
    """Periodically send mock user status changes to a connected client.

    Background task that continuously generates random user status updates
    and sends them to the specified WebSocket connection. Simulates realistic
    user activity by randomly changing user statuses every 4 seconds.

    This method runs until the WebSocket connection is closed or the task
    is cancelled. It selects random users from the current server and updates
    their online status (online, idle, dnd, offline).

    Args:
        websocket: WebSocket connection to send updates to. Must have a
            discordServer attribute identifying which server's users to update.

    Note:
        This is a coroutine that runs indefinitely for the lifetime of the
        WebSocket connection. It automatically handles ConnectionClosed
        exceptions gracefully.

    Examples:

        # Called automatically by the server for each connection
        task = asyncio.create_task(
            mock_provider.periodic_status_updates(websocket)
        )

        # Updates will be sent in this format:
        # {
        #     "type": "presence",
        #     "server": "232769614004748288",
        #     "data": {
        #         "uid": "123456789012345001",
        #         "status": "online"
        #     }
        # }

    Raises:
        websockets.ConnectionClosed: When the WebSocket connection is closed.
            This exception is caught and handled gracefully.
    """
    uids = list(self.get_mock_user_data(websocket.discordServer).keys())
    try:
        while True:
            await asyncio.sleep(4)
            status = self.server._random_status()
            uid = random.choice(uids)
            presence_msg = {
                "type": "presence",
                "server": websocket.discordServer,
                "data": {
                    "uid": uid,
                    "status": status
                }
            }
            print(f"[SEND] presence update for {uid}: {status}")
            await websocket.send(json.dumps(presence_msg))
    except websockets.ConnectionClosed:
        print("[INFO] Presence update task stopped: connection closed")
        # Remove closed connections
        self.server.connections.discard(websocket)
    except Exception as e:
        print(f"[ERROR] Failed to send message to connection: {e}")
        # Optionally remove problematic connections
        self.server.connections.discard(websocket)

periodic_messages(websocket) async

Periodically send mock chat messages to a connected client.

Background task that continuously generates random chat messages from mock users and sends them to the specified WebSocket connection. Simulates realistic chat activity by sending messages every 5 seconds.

This method runs until the WebSocket connection is closed or the task is cancelled. It selects random users from the current server and random messages from a predefined list to create chat events.

Parameters:

Name	Type	Description	Default
websocket	WebSocketServerProtocol	WebSocket connection to send messages to. Must have a discordServer attribute identifying which server's users to use.	required

Note

This is a coroutine that runs indefinitely for the lifetime of the WebSocket connection. It automatically handles ConnectionClosed exceptions gracefully.

Examples:

# Called automatically by the server for each connection
task = asyncio.create_task(
    mock_provider.periodic_messages(websocket)
)

# Messages will be sent in this format:
# {
#     "type": "message",
#     "server": "232769614004748288",
#     "data": {
#         "uid": "123456789012345001",
#         "message": "hello",
#         "channel": "527964146659229701"
#     }
# }

Raises:

Type	Description
ConnectionClosed	When the WebSocket connection is closed. This exception is caught and handled gracefully.

Source code in d_back/mock/data.py

async def periodic_messages(self, websocket: 'WebSocketServerProtocol') -> None:
    """Periodically send mock chat messages to a connected client.

    Background task that continuously generates random chat messages from
    mock users and sends them to the specified WebSocket connection. Simulates
    realistic chat activity by sending messages every 5 seconds.

    This method runs until the WebSocket connection is closed or the task
    is cancelled. It selects random users from the current server and random
    messages from a predefined list to create chat events.

    Args:
        websocket: WebSocket connection to send messages to. Must have a
            discordServer attribute identifying which server's users to use.

    Note:
        This is a coroutine that runs indefinitely for the lifetime of the
        WebSocket connection. It automatically handles ConnectionClosed
        exceptions gracefully.

    Examples:

        # Called automatically by the server for each connection
        task = asyncio.create_task(
            mock_provider.periodic_messages(websocket)
        )

        # Messages will be sent in this format:
        # {
        #     "type": "message",
        #     "server": "232769614004748288",
        #     "data": {
        #         "uid": "123456789012345001",
        #         "message": "hello",
        #         "channel": "527964146659229701"
        #     }
        # }

    Raises:
        websockets.ConnectionClosed: When the WebSocket connection is closed.
            This exception is caught and handled gracefully.
    """
    uids = list(self.get_mock_user_data(websocket.discordServer).keys())
    messages = [
        "hello",
        "how are you?",
        "this is a test message",
        "D-Zone rocks!",
        "what's up?"
    ]
    try:
        while True:
            await asyncio.sleep(5)
            uid = random.choice(uids)
            msg_text = random.choice(messages)
            msg = {
                "type": "message",
                "server": websocket.discordServer,
                "data": {
                    "uid": uid,
                    "message": msg_text,
                    "channel": "527964146659229701"
                }
            }
            print(f"[SEND] periodic message from {uid}: {msg_text}")
            await websocket.send(json.dumps(msg))
    except websockets.ConnectionClosed:
        print("[INFO] Periodic message task stopped: connection closed")
        # Remove closed connections
        self.server.connections.discard(websocket)
    except Exception as e:
        print(f"[ERROR] Failed to send message to connection: {e}")
        # Optionally remove problematic connections
        self.server.connections.discard(websocket)

Utility Functions

Helper functions for command-line interface and version management.

parse_args

Parse command-line arguments for the D-Back WebSocket server.

Provides command-line interface for configuring server parameters including port, host, static file directory, and version information.

Returns:

Type	Description
	argparse.Namespace: Parsed arguments containing: - port (int): Server port number (default: 3000) - host (str): Server hostname (default: 'localhost') - static_dir (str): Custom static files directory (default: None)

Examples:

$ python -m d_back --port 8080 --host 0.0.0.0
$ python -m d_back --static-dir ./my-static-files
$ python -m d_back --version

Source code in d_back/server.py

def parse_args():
    """Parse command-line arguments for the D-Back WebSocket server.

    Provides command-line interface for configuring server parameters including
    port, host, static file directory, and version information.

    Returns:
        argparse.Namespace: Parsed arguments containing:
            - port (int): Server port number (default: 3000)
            - host (str): Server hostname (default: 'localhost')
            - static_dir (str): Custom static files directory (default: None)

    Examples:

        $ python -m d_back --port 8080 --host 0.0.0.0
        $ python -m d_back --static-dir ./my-static-files
        $ python -m d_back --version
    """
    parser = argparse.ArgumentParser(
        description='D-Back WebSocket Server',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter
    )
    parser.add_argument(
        '--port', 
        type=int, 
        default=3000, 
        help='Port to run the WebSocket server on'
    )
    parser.add_argument(
        '--host', 
        type=str, 
        default='localhost', 
        help='Host to bind the WebSocket server to'
    )
    parser.add_argument(
        '--static-dir',
        type=str,
        default=None,
        help='Directory to serve static files from (default: built-in dist directory)'
    )
    parser.add_argument(
        '--version',
        action='version',
        version=f'%(prog)s {get_version()}'
    )
    return parser.parse_args()

get_version

Get the current version of the d_back package.

Attempts to retrieve the version from the package's version attribute. Falls back to "unknown" if the version cannot be determined.

Returns:

Name	Type	Description
str		The package version string (e.g., "0.0.12") or "unknown".

Examples:

>>> get_version()
'0.0.12'

Source code in d_back/server.py

def get_version():
    """Get the current version of the d_back package.

    Attempts to retrieve the version from the package's __version__ attribute.
    Falls back to "unknown" if the version cannot be determined.

    Returns:
        str: The package version string (e.g., "0.0.12") or "unknown".

    Examples:

        >>> get_version()
        '0.0.12'
    """
    try:
        from . import __version__
        return __version__
    except ImportError:
        return "unknown"

main

Main async entry point for the D-Back WebSocket server.

Parses command-line arguments, initializes the WebSocket server with the specified configuration, and starts the server in run-forever mode.

This function handles

• Argument parsing for port, host, and static directory
• Server initialization and configuration
• Static directory validation
• Server startup and lifecycle management

Returns:

Type	Description
	None

Raises:

Type	Description
Exception	If server fails to start or encounters fatal errors.
KeyboardInterrupt	Propagated from server interruption.

Examples:

# Run with default settings
await main()

Note

This is the primary entry point when running d_back as a module. Use main_sync() for synchronous execution from main.

Source code in d_back/server.py

async def main():
    """Main async entry point for the D-Back WebSocket server.

    Parses command-line arguments, initializes the WebSocket server with the
    specified configuration, and starts the server in run-forever mode.

    This function handles:
        - Argument parsing for port, host, and static directory
        - Server initialization and configuration
        - Static directory validation
        - Server startup and lifecycle management

    Returns:
        None

    Raises:
        Exception: If server fails to start or encounters fatal errors.
        KeyboardInterrupt: Propagated from server interruption.

    Examples:

        # Run with default settings
        await main()

    Note:
        This is the primary entry point when running d_back as a module.
        Use main_sync() for synchronous execution from __main__.
    """
    args = parse_args()

    print(f"Starting D-Back WebSocket Server v{get_version()}")
    print(f"Host: {args.host}")
    print(f"Port: {args.port}")

    server = WebSocketServer(port=args.port, host=args.host)

    # Set custom static directory if provided
    if args.static_dir:
        static_path = Path(args.static_dir)
        if static_path.exists() and static_path.is_dir():
            server.static_dir = static_path
            print(f"Static directory: {static_path}")
        else:
            print(f"Warning: Static directory '{args.static_dir}' does not exist or is not a directory")
            print(f"Using default static directory: {server.static_dir}")
    else:
        print(f"Static directory: {server.static_dir}")

    await server.run_forever()

main_sync

Synchronous entry point wrapper for the D-Back WebSocket server.

Wraps the async main() function in asyncio.run() to provide a synchronous entry point. Handles KeyboardInterrupt gracefully for clean server shutdown.

Returns:

Type	Description
	None

Examples:

if __name__ == "__main__":
    main_sync()

Note

This is the entry point used when running as a script or via setuptools console_scripts. It ensures proper async context management.

Source code in d_back/server.py

def main_sync():
    """Synchronous entry point wrapper for the D-Back WebSocket server.

    Wraps the async main() function in asyncio.run() to provide a synchronous
    entry point. Handles KeyboardInterrupt gracefully for clean server shutdown.

    Returns:
        None

    Examples:

        if __name__ == "__main__":
            main_sync()

    Note:
        This is the entry point used when running as a script or via setuptools
        console_scripts. It ensures proper async context management.
    """
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n[INFO] Server stopped by user")

Usage Examples

For practical examples of using these APIs, see:

• Getting Started: Basic usage and first connection
• Configuration: Server setup and configuration
• Callbacks & Customization: Callback usage examples
• Custom Data Providers: Data provider implementation patterns

Type Hints

All methods include comprehensive type hints for parameters and return values. When working with callbacks, import the necessary types:

from typing import Dict, Any, Optional, Tuple, Callable, Awaitable

For more information about Python type hints, see the official typing documentation.