Exceptions¶

All library exceptions inherit from MixpanelDataError, enabling callers to catch all library errors with a single except clause.

Explore on DeepWiki

🤖 Error Handling Guide →

Ask questions about specific exceptions, error recovery patterns, or debugging strategies.

Exception Hierarchy¶

MixpanelDataError
├── ConfigError
│   ├── AccountNotFoundError
│   └── AccountExistsError
├── APIError
│   ├── AuthenticationError
│   ├── RateLimitError
│   ├── QueryError
│   ├── ServerError
│   └── JQLSyntaxError
├── TableExistsError
├── TableNotFoundError
├── DatabaseLockedError
└── DatabaseNotFoundError

Catching Errors¶

import mixpanel_data as mp

try:
    ws = mp.Workspace()
    result = ws.segmentation(event="Purchase", from_date="2025-01-01", to_date="2025-01-31")
except mp.AuthenticationError as e:
    print(f"Auth failed: {e.message}")
except mp.RateLimitError as e:
    print(f"Rate limited, retry after {e.retry_after}s")
except mp.MixpanelDataError as e:
    print(f"Error [{e.code}]: {e.message}")

Base Exception¶

mixpanel_data.MixpanelDataError ¶

MixpanelDataError(
    message: str,
    code: str = "UNKNOWN_ERROR",
    details: dict[str, Any] | None = None,
)

Bases: Exception

Base exception for all mixpanel_data errors.

All library exceptions inherit from this class, allowing callers to: - Catch all library errors: except MixpanelDataError - Handle specific errors: except AccountNotFoundError - Serialize errors: error.to_dict()

Initialize exception.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str`
`code`	Machine-readable error code for programmatic handling. TYPE: `str` DEFAULT: `'UNKNOWN_ERROR'`
`details`	Additional structured data about the error. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str,
    code: str = "UNKNOWN_ERROR",
    details: dict[str, Any] | None = None,
) -> None:
    """Initialize exception.

    Args:
        message: Human-readable error message.
        code: Machine-readable error code for programmatic handling.
        details: Additional structured data about the error.
    """
    super().__init__(message)
    self._message = message
    self._code = code
    self._details = details or {}

code `property` ¶

code: str

Machine-readable error code.

message `property` ¶

message: str

Human-readable error message.

details `property` ¶

details: dict[str, Any]

Additional structured error data.

to_dict ¶

to_dict() -> dict[str, Any]

Serialize exception for logging/JSON output.

RETURNS	DESCRIPTION
`dict[str, Any]`	Dictionary with keys: code, message, details.
`dict[str, Any]`	All values are JSON-serializable.

Source code in src/mixpanel_data/exceptions.py

def to_dict(self) -> dict[str, Any]:
    """Serialize exception for logging/JSON output.

    Returns:
        Dictionary with keys: code, message, details.
        All values are JSON-serializable.
    """
    return {
        "code": self._code,
        "message": self._message,
        "details": self._details,
    }

str ¶

__str__() -> str

Return human-readable error message.

Source code in src/mixpanel_data/exceptions.py

def __str__(self) -> str:
    """Return human-readable error message."""
    return self._message

repr ¶

__repr__() -> str

Return detailed string representation.

Source code in src/mixpanel_data/exceptions.py

def __repr__(self) -> str:
    """Return detailed string representation."""
    return (
        f"{self.__class__.__name__}(message={self._message!r}, code={self._code!r})"
    )

API Exceptions¶

mixpanel_data.APIError ¶

APIError(
    message: str,
    *,
    status_code: int,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
    request_body: dict[str, Any] | None = None,
    code: str = "API_ERROR",
)

Bases: MixpanelDataError

Base class for Mixpanel API HTTP errors.

Provides structured access to HTTP request/response context for debugging and automated recovery by AI agents. All API-related exceptions inherit from this class, enabling agents to:

Understand what went wrong (status code, error message)
See exactly what was sent (request method, URL, params, body)
See exactly what came back (response body, headers)
Modify their approach and retry autonomously

Example

try:
    result = client.segmentation(event="signup", ...)
except APIError as e:
    print(f"Status: {e.status_code}")
    print(f"Response: {e.response_body}")
    print(f"Request URL: {e.request_url}")
    print(f"Request params: {e.request_params}")

Initialize APIError.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str`
`status_code`	HTTP status code from response. TYPE: `int`
`response_body`	Raw response body (string or parsed dict). TYPE: `str \| dict[str, Any] \| None` DEFAULT: `None`
`request_method`	HTTP method used (GET, POST). TYPE: `str \| None` DEFAULT: `None`
`request_url`	Full request URL. TYPE: `str \| None` DEFAULT: `None`
`request_params`	Query parameters sent. TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`request_body`	Request body sent (for POST requests). TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`code`	Machine-readable error code. TYPE: `str` DEFAULT: `'API_ERROR'`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str,
    *,
    status_code: int,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
    request_body: dict[str, Any] | None = None,
    code: str = "API_ERROR",
) -> None:
    """Initialize APIError.

    Args:
        message: Human-readable error message.
        status_code: HTTP status code from response.
        response_body: Raw response body (string or parsed dict).
        request_method: HTTP method used (GET, POST).
        request_url: Full request URL.
        request_params: Query parameters sent.
        request_body: Request body sent (for POST requests).
        code: Machine-readable error code.
    """
    self._status_code = status_code
    self._response_body = response_body
    self._request_method = request_method
    self._request_url = request_url
    self._request_params = request_params
    self._request_body = request_body

    details: dict[str, Any] = {
        "status_code": status_code,
    }
    if response_body is not None:
        details["response_body"] = response_body
    if request_method is not None:
        details["request_method"] = request_method
    if request_url is not None:
        details["request_url"] = request_url
    if request_params is not None:
        details["request_params"] = request_params
    if request_body is not None:
        details["request_body"] = request_body

    super().__init__(message, code=code, details=details)

status_code `property` ¶

status_code: int

HTTP status code from response.

response_body `property` ¶

response_body: str | dict[str, Any] | None

Raw response body (string or parsed dict).

request_method `property` ¶

request_method: str | None

HTTP method used (GET, POST).

request_url `property` ¶

request_url: str | None

Full request URL.

request_params `property` ¶

request_params: dict[str, Any] | None

Query parameters sent.

request_body `property` ¶

request_body: dict[str, Any] | None

Request body sent (for POST requests).

mixpanel_data.AuthenticationError ¶

AuthenticationError(
    message: str = "Authentication failed",
    *,
    status_code: int = 401,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
)

Bases: APIError

Authentication with Mixpanel API failed (HTTP 401).

Raised when credentials are invalid, expired, or lack required permissions. Inherits from APIError to provide full request/response context.

Example

try:
    client.segmentation(...)
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
    print(f"Request URL: {e.request_url}")
    # Check if project_id is correct, credentials are valid, etc.

Initialize AuthenticationError.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str` DEFAULT: `'Authentication failed'`
`status_code`	HTTP status code (default 401). TYPE: `int` DEFAULT: `401`
`response_body`	Raw response body. TYPE: `str \| dict[str, Any] \| None` DEFAULT: `None`
`request_method`	HTTP method used. TYPE: `str \| None` DEFAULT: `None`
`request_url`	Full request URL. TYPE: `str \| None` DEFAULT: `None`
`request_params`	Query parameters sent. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str = "Authentication failed",
    *,
    status_code: int = 401,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
) -> None:
    """Initialize AuthenticationError.

    Args:
        message: Human-readable error message.
        status_code: HTTP status code (default 401).
        response_body: Raw response body.
        request_method: HTTP method used.
        request_url: Full request URL.
        request_params: Query parameters sent.
    """
    super().__init__(
        message,
        status_code=status_code,
        response_body=response_body,
        request_method=request_method,
        request_url=request_url,
        request_params=request_params,
        code="AUTH_FAILED",
    )

mixpanel_data.RateLimitError ¶

RateLimitError(
    message: str = "Rate limit exceeded",
    *,
    retry_after: int | None = None,
    status_code: int = 429,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
)

Bases: APIError

Mixpanel API rate limit exceeded (HTTP 429).

Raised when the API returns a 429 status. The retry_after property indicates when the request can be retried. Inherits from APIError to provide full request context for debugging.

Example

try:
    for _ in range(1000):
        client.segmentation(...)
except RateLimitError as e:
    print(f"Rate limited! Retry after {e.retry_after}s")
    print(f"Request: {e.request_method} {e.request_url}")
    time.sleep(e.retry_after or 60)

Initialize RateLimitError.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str` DEFAULT: `'Rate limit exceeded'`
`retry_after`	Seconds until retry is allowed (from Retry-After header). TYPE: `int \| None` DEFAULT: `None`
`status_code`	HTTP status code (default 429). TYPE: `int` DEFAULT: `429`
`response_body`	Raw response body. TYPE: `str \| dict[str, Any] \| None` DEFAULT: `None`
`request_method`	HTTP method used. TYPE: `str \| None` DEFAULT: `None`
`request_url`	Full request URL. TYPE: `str \| None` DEFAULT: `None`
`request_params`	Query parameters sent. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str = "Rate limit exceeded",
    *,
    retry_after: int | None = None,
    status_code: int = 429,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
) -> None:
    """Initialize RateLimitError.

    Args:
        message: Human-readable error message.
        retry_after: Seconds until retry is allowed (from Retry-After header).
        status_code: HTTP status code (default 429).
        response_body: Raw response body.
        request_method: HTTP method used.
        request_url: Full request URL.
        request_params: Query parameters sent.
    """
    self._retry_after = retry_after
    if retry_after is not None:
        message = f"{message}. Retry after {retry_after} seconds."

    super().__init__(
        message,
        status_code=status_code,
        response_body=response_body,
        request_method=request_method,
        request_url=request_url,
        request_params=request_params,
        code="RATE_LIMITED",
    )
    # Add retry_after to details
    if retry_after is not None:
        self._details["retry_after"] = retry_after

retry_after `property` ¶

retry_after: int | None

Seconds until retry is allowed, or None if unknown.

mixpanel_data.QueryError ¶

QueryError(
    message: str = "Query execution failed",
    *,
    status_code: int = 400,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
    request_body: dict[str, Any] | None = None,
)

Bases: APIError

Query execution failed (HTTP 400 or query-specific error).

Raised when an API query fails due to invalid parameters, syntax errors, or other query-specific issues. Inherits from APIError to provide full request/response context for debugging.

Example

try:
    client.segmentation(event="nonexistent", ...)
except QueryError as e:
    print(f"Query failed: {e.message}")
    print(f"Response: {e.response_body}")
    print(f"Request params: {e.request_params}")

Initialize QueryError.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str` DEFAULT: `'Query execution failed'`
`status_code`	HTTP status code (default 400). TYPE: `int` DEFAULT: `400`
`response_body`	Raw response body with error details. TYPE: `str \| dict[str, Any] \| None` DEFAULT: `None`
`request_method`	HTTP method used. TYPE: `str \| None` DEFAULT: `None`
`request_url`	Full request URL. TYPE: `str \| None` DEFAULT: `None`
`request_params`	Query parameters sent. TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`request_body`	Request body sent (for POST). TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str = "Query execution failed",
    *,
    status_code: int = 400,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
    request_body: dict[str, Any] | None = None,
) -> None:
    """Initialize QueryError.

    Args:
        message: Human-readable error message.
        status_code: HTTP status code (default 400).
        response_body: Raw response body with error details.
        request_method: HTTP method used.
        request_url: Full request URL.
        request_params: Query parameters sent.
        request_body: Request body sent (for POST).
    """
    super().__init__(
        message,
        status_code=status_code,
        response_body=response_body,
        request_method=request_method,
        request_url=request_url,
        request_params=request_params,
        request_body=request_body,
        code="QUERY_FAILED",
    )

mixpanel_data.ServerError ¶

ServerError(
    message: str = "Server error",
    *,
    status_code: int = 500,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
    request_body: dict[str, Any] | None = None,
)

Bases: APIError

Mixpanel server error (HTTP 5xx).

Raised when the Mixpanel API returns a server error. These are typically transient issues that may succeed on retry. The response_body property contains the full error details from Mixpanel, which often include actionable information (e.g., "unit and interval both specified").

Example

try:
    client.retention(born_event="signup", ...)
except ServerError as e:
    print(f"Server error {e.status_code}: {e.message}")
    print(f"Response: {e.response_body}")
    print(f"Request params: {e.request_params}")
    # AI agent can analyze response_body to fix the request

Initialize ServerError.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str` DEFAULT: `'Server error'`
`status_code`	HTTP status code (5xx). TYPE: `int` DEFAULT: `500`
`response_body`	Raw response body with error details. TYPE: `str \| dict[str, Any] \| None` DEFAULT: `None`
`request_method`	HTTP method used. TYPE: `str \| None` DEFAULT: `None`
`request_url`	Full request URL. TYPE: `str \| None` DEFAULT: `None`
`request_params`	Query parameters sent. TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`request_body`	Request body sent (for POST). TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str = "Server error",
    *,
    status_code: int = 500,
    response_body: str | dict[str, Any] | None = None,
    request_method: str | None = None,
    request_url: str | None = None,
    request_params: dict[str, Any] | None = None,
    request_body: dict[str, Any] | None = None,
) -> None:
    """Initialize ServerError.

    Args:
        message: Human-readable error message.
        status_code: HTTP status code (5xx).
        response_body: Raw response body with error details.
        request_method: HTTP method used.
        request_url: Full request URL.
        request_params: Query parameters sent.
        request_body: Request body sent (for POST).
    """
    super().__init__(
        message,
        status_code=status_code,
        response_body=response_body,
        request_method=request_method,
        request_url=request_url,
        request_params=request_params,
        request_body=request_body,
        code="SERVER_ERROR",
    )

mixpanel_data.JQLSyntaxError ¶

JQLSyntaxError(
    raw_error: str, script: str | None = None, request_path: str | None = None
)

Bases: QueryError

JQL script execution failed with syntax or runtime error (HTTP 412).

Raised when a JQL script fails to execute due to syntax errors, type errors, or other JavaScript runtime issues. Provides structured access to error details from Mixpanel's response.

Inherits from QueryError (and thus APIError) to provide full HTTP context.

Example

try:
    result = live_query.jql(script)
except JQLSyntaxError as e:
    print(f"Error: {e.error_type}: {e.error_message}")
    print(f"Script: {e.script}")
    print(f"Line info: {e.line_info}")
    # AI agent can use this to fix the script and retry

Initialize JQLSyntaxError.

PARAMETER	DESCRIPTION
`raw_error`	Raw error string from Mixpanel API response. TYPE: `str`
`script`	The JQL script that caused the error. TYPE: `str \| None` DEFAULT: `None`
`request_path`	API request path from error response. TYPE: `str \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    raw_error: str,
    script: str | None = None,
    request_path: str | None = None,
) -> None:
    """Initialize JQLSyntaxError.

    Args:
        raw_error: Raw error string from Mixpanel API response.
        script: The JQL script that caused the error.
        request_path: API request path from error response.
    """
    # Parse structured error info from raw error string
    self._error_type = self._extract_error_type(raw_error)
    self._error_message = self._extract_message(raw_error)
    self._line_info = self._extract_line_info(raw_error)
    self._stack_trace = self._extract_stack_trace(raw_error)
    self._script = script
    self._raw_error = raw_error
    self._request_path = request_path

    # Build human-readable message
    message = f"JQL {self._error_type}: {self._error_message}"
    if self._line_info:
        message += f"\n{self._line_info}"

    # Build response body dict for APIError
    response_body: dict[str, Any] = {
        "error": raw_error,
    }
    if request_path:
        response_body["request"] = request_path

    super().__init__(
        message,
        status_code=412,
        response_body=response_body,
        request_body={"script": script} if script else None,
    )
    self._code = "JQL_SYNTAX_ERROR"

    # Add JQL-specific details
    self._details["error_type"] = self._error_type
    self._details["error_message"] = self._error_message
    self._details["line_info"] = self._line_info
    self._details["stack_trace"] = self._stack_trace
    self._details["script"] = script
    self._details["request_path"] = request_path
    self._details["raw_error"] = raw_error

error_type `property` ¶

error_type: str

JavaScript error type (TypeError, SyntaxError, ReferenceError, etc.).

error_message `property` ¶

error_message: str

Error message describing what went wrong.

line_info `property` ¶

line_info: str | None

Code snippet with caret showing error location, if available.

stack_trace `property` ¶

stack_trace: str | None

JavaScript stack trace, if available.

script `property` ¶

script: str | None

The JQL script that caused the error.

raw_error `property` ¶

raw_error: str

Complete raw error string from Mixpanel.

Configuration Exceptions¶

mixpanel_data.ConfigError ¶

ConfigError(message: str, details: dict[str, Any] | None = None)

Bases: MixpanelDataError

Base for configuration-related errors.

Raised when there's a problem with configuration files, environment variables, or credential resolution.

Initialize ConfigError.

PARAMETER	DESCRIPTION
`message`	Human-readable error message. TYPE: `str`
`details`	Additional structured data. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    message: str,
    details: dict[str, Any] | None = None,
) -> None:
    """Initialize ConfigError.

    Args:
        message: Human-readable error message.
        details: Additional structured data.
    """
    super().__init__(message, code="CONFIG_ERROR", details=details)

mixpanel_data.AccountNotFoundError ¶

AccountNotFoundError(
    account_name: str, available_accounts: list[str] | None = None
)

Bases: ConfigError

Named account does not exist in configuration.

Raised when attempting to access an account that hasn't been configured. The available_accounts property lists valid account names to help users.

Initialize AccountNotFoundError.

PARAMETER	DESCRIPTION
`account_name`	The requested account name that wasn't found. TYPE: `str`
`available_accounts`	List of valid account names for suggestions. TYPE: `list[str] \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    account_name: str,
    available_accounts: list[str] | None = None,
) -> None:
    """Initialize AccountNotFoundError.

    Args:
        account_name: The requested account name that wasn't found.
        available_accounts: List of valid account names for suggestions.
    """
    available = available_accounts or []
    if available:
        available_str = ", ".join(f"'{a}'" for a in available)
        message = (
            f"Account '{account_name}' not found. "
            f"Available accounts: {available_str}"
        )
    else:
        message = f"Account '{account_name}' not found. No accounts configured."

    details = {
        "account_name": account_name,
        "available_accounts": available,
    }
    super().__init__(message, details=details)
    self._code = "ACCOUNT_NOT_FOUND"

account_name `property` ¶

account_name: str

The requested account name that wasn't found.

available_accounts `property` ¶

available_accounts: list[str]

List of valid account names.

mixpanel_data.AccountExistsError ¶

AccountExistsError(account_name: str)

Bases: ConfigError

Account name already exists in configuration.

Raised when attempting to add an account with a name that's already in use.

Initialize AccountExistsError.

PARAMETER	DESCRIPTION
`account_name`	The conflicting account name. TYPE: `str`

Source code in src/mixpanel_data/exceptions.py

def __init__(self, account_name: str) -> None:
    """Initialize AccountExistsError.

    Args:
        account_name: The conflicting account name.
    """
    message = f"Account '{account_name}' already exists."
    details = {"account_name": account_name}
    super().__init__(message, details=details)
    self._code = "ACCOUNT_EXISTS"

account_name `property` ¶

account_name: str

The conflicting account name.

Storage Exceptions¶

Storage exceptions are raised during fetch and table operations:

Exception	Raised When
`TableExistsError`	Fetching to an existing table without `append=True` or `--replace`
`TableNotFoundError`	Using `append=True` on a non-existent table
`DatabaseLockedError`	Another process has the database locked
`DatabaseNotFoundError`	Database file not found in read-only mode

mixpanel_data.TableExistsError ¶

TableExistsError(table_name: str)

Bases: MixpanelDataError

Table already exists in local database.

Raised when attempting to create a table that already exists. Use drop() first to remove the existing table.

Initialize TableExistsError.

PARAMETER	DESCRIPTION
`table_name`	Name of the existing table. TYPE: `str`

Source code in src/mixpanel_data/exceptions.py

def __init__(self, table_name: str) -> None:
    """Initialize TableExistsError.

    Args:
        table_name: Name of the existing table.
    """
    message = f"Table '{table_name}' already exists."
    details = {
        "table_name": table_name,
        "suggestion": "Use drop() first to remove the existing table.",
    }
    super().__init__(message, code="TABLE_EXISTS", details=details)

table_name `property` ¶

table_name: str

Name of the existing table.

mixpanel_data.TableNotFoundError ¶

TableNotFoundError(table_name: str)

Bases: MixpanelDataError

Table does not exist in local database.

Raised when attempting to access a table that hasn't been created.

Initialize TableNotFoundError.

PARAMETER	DESCRIPTION
`table_name`	Name of the missing table. TYPE: `str`

Source code in src/mixpanel_data/exceptions.py

def __init__(self, table_name: str) -> None:
    """Initialize TableNotFoundError.

    Args:
        table_name: Name of the missing table.
    """
    message = f"Table '{table_name}' not found."
    details = {"table_name": table_name}
    super().__init__(message, code="TABLE_NOT_FOUND", details=details)

table_name `property` ¶

table_name: str

Name of the missing table.

mixpanel_data.DatabaseLockedError ¶

DatabaseLockedError(db_path: str, holding_pid: int | None = None)

Bases: MixpanelDataError

Database is locked by another process.

Raised when attempting to access a DuckDB database that is locked by another process. DuckDB uses single-writer, multiple-reader concurrency - only one process can have write access at a time.

Example

try:
    ws = Workspace()
except DatabaseLockedError as e:
    print(f"Database {e.db_path} is locked")
    if e.holding_pid:
        print(f"Held by PID {e.holding_pid}")

Initialize DatabaseLockedError.

PARAMETER	DESCRIPTION
`db_path`	Path to the locked database file. TYPE: `str`
`holding_pid`	Process ID holding the lock, if available. TYPE: `int \| None` DEFAULT: `None`

Source code in src/mixpanel_data/exceptions.py

def __init__(
    self,
    db_path: str,
    holding_pid: int | None = None,
) -> None:
    """Initialize DatabaseLockedError.

    Args:
        db_path: Path to the locked database file.
        holding_pid: Process ID holding the lock, if available.
    """
    message = f"Database '{db_path}' is locked by another process"
    if holding_pid is not None:
        message += f" (PID {holding_pid})"
    message += ". Wait for the other operation to complete and try again."

    details: dict[str, str | int] = {
        "db_path": db_path,
        "suggestion": "Wait for the other operation to complete and try again.",
    }
    if holding_pid is not None:
        details["holding_pid"] = holding_pid

    super().__init__(message, code="DATABASE_LOCKED", details=details)

db_path `property` ¶

db_path: str

Path to the locked database.

holding_pid `property` ¶

holding_pid: int | None

Process ID holding the lock, if available.

mixpanel_data.DatabaseNotFoundError ¶

DatabaseNotFoundError(db_path: str)

Bases: MixpanelDataError

Database file does not exist.

Raised when attempting to open a non-existent database file in read-only mode. DuckDB cannot create a new database file when opened read-only.

This typically happens when running read-only commands (like mp query or mp inspect tables) before any data has been fetched.

Example

try:
    ws = Workspace(read_only=True)
except DatabaseNotFoundError as e:
    print(f"No data yet: {e.db_path}")
    print("Run 'mp fetch events' first to create the database.")

Initialize DatabaseNotFoundError.

PARAMETER	DESCRIPTION
`db_path`	Path to the database file that doesn't exist. TYPE: `str`

Source code in src/mixpanel_data/exceptions.py

def __init__(self, db_path: str) -> None:
    """Initialize DatabaseNotFoundError.

    Args:
        db_path: Path to the database file that doesn't exist.
    """
    message = (
        f"Database '{db_path}' does not exist. "
        "Run 'mp fetch events' first to create it."
    )

    details: dict[str, str] = {
        "db_path": db_path,
        "suggestion": "Run 'mp fetch events' or 'mp fetch profiles' to create the database.",
    }

    super().__init__(message, code="DATABASE_NOT_FOUND", details=details)

db_path `property` ¶

db_path: str

Path to the database file that doesn't exist.

Exceptions¶

Exception Hierarchy¶

Catching Errors¶

Base Exception¶

mixpanel_data.MixpanelDataError ¶

code property ¶

message property ¶

details property ¶

to_dict ¶

__str__ ¶

__repr__ ¶

API Exceptions¶

mixpanel_data.APIError ¶

status_code property ¶

response_body property ¶

request_method property ¶

request_url property ¶

request_params property ¶

request_body property ¶

mixpanel_data.AuthenticationError ¶

mixpanel_data.RateLimitError ¶

retry_after property ¶

mixpanel_data.QueryError ¶

mixpanel_data.ServerError ¶

mixpanel_data.JQLSyntaxError ¶

error_type property ¶

error_message property ¶

line_info property ¶

stack_trace property ¶

script property ¶

raw_error property ¶

Configuration Exceptions¶

mixpanel_data.ConfigError ¶

mixpanel_data.AccountNotFoundError ¶

account_name property ¶

available_accounts property ¶

mixpanel_data.AccountExistsError ¶

account_name property ¶

Storage Exceptions¶

mixpanel_data.TableExistsError ¶

table_name property ¶

mixpanel_data.TableNotFoundError ¶

table_name property ¶

mixpanel_data.DatabaseLockedError ¶

db_path property ¶

holding_pid property ¶

mixpanel_data.DatabaseNotFoundError ¶

db_path property ¶

code `property` ¶

message `property` ¶

details `property` ¶

str ¶

repr ¶

status_code `property` ¶

response_body `property` ¶

request_method `property` ¶

request_url `property` ¶

request_params `property` ¶

request_body `property` ¶

retry_after `property` ¶

error_type `property` ¶

error_message `property` ¶

line_info `property` ¶

stack_trace `property` ¶

script `property` ¶

raw_error `property` ¶

account_name `property` ¶

available_accounts `property` ¶

account_name `property` ¶

table_name `property` ¶

table_name `property` ¶

db_path `property` ¶

holding_pid `property` ¶

db_path `property` ¶