Client API

This section documents the main client classes and their methods.

Main VNDB Client

class veedb.VNDB(api_token: str | None = None, use_sandbox: bool = False, session: ClientSession | None = None, local_schema_path: str | None = None, schema_cache_dir: str = '.veedb_cache', schema_cache_ttl_hours: float = 360, base_url: str | None = None)

Bases: object

__init__(api_token: str | None = None, use_sandbox: bool = False, session: ClientSession | None = None, local_schema_path: str | None = None, schema_cache_dir: str = '.veedb_cache', schema_cache_ttl_hours: float = 360, base_url: str | None = None)

Parameters:

base_url – Override the kana API endpoint. Useful for self-hosted replicas. Falls back to the VEEDB_BASE_URL environment variable, then to the upstream api.vndb.org/kana (or sandbox if use_sandbox=True). Strip any trailing slash.

The main client class for interacting with the VNDB API.

Parameters:

• api_token (Optional[str]): API token for authenticated requests

• use_sandbox (bool): Whether to use the sandbox API endpoint (default: False)

• session (Optional[aiohttp.ClientSession]): Custom aiohttp session

• local_schema_path (Optional[str]): Path to local schema file

• schema_cache_dir (str): Directory for schema cache (default: “.veedb_cache”)

• schema_cache_ttl_hours (float): Cache TTL in hours (default: 24.0)

Basic Usage:

async with VNDB() as client:
    stats = await client.get_stats()
    print(f"Total VNs: {stats.vn}")

With Authentication:

async with VNDB(api_token="your-token") as client:
    auth_info = await client.get_authinfo()
    print(f"Logged in as: {auth_info.username}")

__init__(api_token: str | None = None, use_sandbox: bool = False, session: ClientSession | None = None, local_schema_path: str | None = None, schema_cache_dir: str = '.veedb_cache', schema_cache_ttl_hours: float = 360, base_url: str | None = None)

Parameters:

base_url – Override the kana API endpoint. Useful for self-hosted replicas. Falls back to the VEEDB_BASE_URL environment variable, then to the upstream api.vndb.org/kana (or sandbox if use_sandbox=True). Strip any trailing slash.

async close()

async get_schema() → dict

Get the VNDB API schema, using cache if available and not older than configured TTL. Downloads and caches the schema if cache doesn’t exist or is expired. Uses the same schema cache as the filter validation system.

async get_enums() → Dict[str, Any]

Get enum definitions from the VNDB API schema (uses shared schema cache).

async get_stats() → UserStats

async get_user(q: str | List[str], fields: str | None = None) → Dict[str, User | None]

async get_authinfo(token: str = None) → AuthInfo

async validate_filters(endpoint: str, filters: List | str | None) → Dict[str, Any]

Validates filters against the schema for a specific endpoint.

async get_available_fields(endpoint: str) → List[str]

Get all available filterable fields for an endpoint.

async list_endpoints() → List[str]

Get all available API endpoints.

invalidate_schema_cache()

Invalidates the schema cache, forcing a refresh on next validation or schema access.

async update_local_schema() → Dict[str, Any]

Forces a download of the latest schema and updates the local schema file (if configured) or the cache.

Endpoint-Specific Clients

Visual Novel Client

class veedb.client._VNClient(client: VNDB)

Bases: _BaseEntityClient[VN, VN]

Accessed via client.vn. Provides methods for querying visual novel data.

Example:

query = QueryRequest(
    filters=["title", "~", "fate"],
    fields="id, title, rating"
)
response = await client.vn.query(query)

__init__(client: VNDB)

Character Client

class veedb.client._CharacterClient(client: VNDB)

Bases: _BaseEntityClient[Character, Character]

Accessed via client.character. Provides methods for querying character data.

__init__(client: VNDB)

Release Client

class veedb.client._ReleaseClient(client: VNDB)

Bases: _BaseEntityClient[Release, Release]

Accessed via client.release. Provides methods for querying release data.

__init__(client: VNDB)

Producer Client

class veedb.client._ProducerClient(client: VNDB)

Bases: _BaseEntityClient[Producer, Producer]

Accessed via client.producer. Provides methods for querying producer data.

__init__(client: VNDB)

Staff Client

class veedb.client._StaffClient(client: VNDB)

Bases: _BaseEntityClient[Staff, Staff]

Accessed via client.staff. Provides methods for querying staff data.

__init__(client: VNDB)

Tag Client

class veedb.client._TagClient(client: VNDB)

Bases: _BaseEntityClient[Tag, Tag]

Accessed via client.tag. Provides methods for querying tag data.

__init__(client: VNDB)

Trait Client

class veedb.client._TraitClient(client: VNDB)

Bases: _BaseEntityClient[Trait, Trait]

Accessed via client.trait. Provides methods for querying trait data.

__init__(client: VNDB)

Quote Client

class veedb.client._QuoteClient(client: VNDB)

Bases: _BaseEntityClient[Quote, Quote]

Accessed via client.quote. Provides methods for querying quote data.

__init__(client: VNDB)

User List Client

class veedb.client._UlistClient(client: VNDB)

Bases: object

Accessed via client.ulist. Provides methods for managing user visual novel lists.

Requires authentication for write operations.

Example:

# Query user's list
query = QueryRequest(
    filters=["uid", "=", user_id],
    fields="id, vote, notes, vn{title}"
)
response = await client.ulist.query(query)

# Update list entry
payload = UlistUpdatePayload(id="v17", vote=85, notes="Great VN!")
await client.ulist.update("v17", payload)

__init__(client: VNDB)

async query(user_id: str | None = None, query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False)) → QueryResponse[UlistItem]

async get_labels(user_id: str | None = None, fields: str | None = None) → List[UlistLabel]

async update_entry(vn_id: str, payload: UlistUpdatePayload) → None

async delete_entry(vn_id: str) → None

async query_all_pages(user_id: str, query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False), max_pages: int | None = None) → List[UlistItem]

Fetch all ulist results across multiple pages automatically.

Parameters:

• user_id – The user ID to query

• query_options – The query to execute

• max_pages – Maximum number of pages to fetch (None for unlimited)

Returns:

List of all results from all pages

async query_paginated(user_id: str, query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False)) → AsyncGenerator[QueryResponse[UlistItem], None]

Generator that yields ulist query responses page by page.

Parameters:

• user_id – The user ID to query

• query_options – The query to execute

Yields:

QueryResponse objects for each page

Release List Client

class veedb.client._RlistClient(client: VNDB)

Bases: object

Accessed via client.rlist. Provides methods for managing user release lists.

Requires authentication for write operations.

__init__(client: VNDB)

async update_entry(release_id: str, payload: RlistUpdatePayload) → None

async delete_entry(release_id: str) → None

Base Entity Client

class veedb.client._BaseEntityClient(client: VNDB, endpoint_path: str, entity_dataclass: Type[T_Entity], query_item_dataclass: Type[T_QueryItem])

Bases: Generic[T_Entity, T_QueryItem]

Base class for all endpoint-specific clients. Provides common functionality like query methods and filter validation.

__init__(client: VNDB, endpoint_path: str, entity_dataclass: Type[T_Entity], query_item_dataclass: Type[T_QueryItem])

async query(query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False)) → QueryResponse[T_QueryItem]

async query_all_pages(query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False), max_pages: int | None = None) → List[T_QueryItem]

Fetch all results across multiple pages automatically.

Parameters:

• query_options – The query to execute

• max_pages – Maximum number of pages to fetch (None for unlimited)

Returns:

List of all results from all pages

async query_paginated(query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False)) → AsyncGenerator[QueryResponse[T_QueryItem], None]

Generator that yields query responses page by page.

Parameters:

query_options – The query to execute

Yields:

QueryResponse objects for each page

async validate_filters(filters: List | str | None) → Dict[str, Any]

Validates filters against the schema for this specific endpoint.

async get_available_fields() → List[str]

Gets all available filterable fields for this endpoint.

Client Methods Reference

Query Methods

All endpoint clients inherit these methods from _BaseEntityClient:

async _BaseEntityClient.query(query_options: QueryRequest = QueryRequest(filters=[], fields='id', sort='id', reverse=False, results=10, page=1, user=None, count=False, compact_filters=False, normalized_filters=False)) → QueryResponse[T_QueryItem]

Query the endpoint with the given parameters.

Parameters:

• query_options (QueryRequest): Query parameters including filters, fields, sorting, etc.

Returns:

• QueryResponse[T_QueryItem]: Response containing results and metadata

async _BaseEntityClient.validate_filters(filters: List | str | None) → Dict[str, Any]

Validates filters against the schema for this specific endpoint.

Validate filters against the API schema for this endpoint.

Parameters:

• filters (Union[List, str, None]): Filter expression to validate

Returns:

• Dict[str, Any]: Validation result with ‘valid’, ‘errors’, ‘suggestions’ keys

async _BaseEntityClient.get_available_fields() → List[str]

Gets all available filterable fields for this endpoint.

Get all available fields for this endpoint.

Returns:

• List[str]: List of available field names

List Management Methods

For _UlistClient and _RlistClient:

Note

The user list management methods (update, delete) are currently not implemented in the client. Please refer to the main VNDB API documentation for direct API usage.

Main Client Methods

async VNDB.get_stats() → UserStats

Get database statistics.

Returns:

• Stats: Database statistics including VN count, character count, etc.

async VNDB.get_schema() → dict

Get the VNDB API schema, using cache if available and not older than configured TTL. Downloads and caches the schema if cache doesn’t exist or is expired. Uses the same schema cache as the filter validation system.

Get the API schema.

Returns:

• Dict[str, Any]: Complete API schema

async VNDB.get_user(q: str | List[str], fields: str | None = None) → Dict[str, User | None]

Get user information by ID.

Parameters:

• user_id (str): User ID to query

Returns:

• User: User information

async VNDB.get_authinfo(token: str = None) → AuthInfo

Get authentication information for the current token.

Returns:

• AuthInfo: Authentication details

Requires: Authentication

Validation Methods

async VNDB.validate_filters(endpoint: str, filters: List | str | None) → Dict[str, Any]

Validates filters against the schema for a specific endpoint.

Validate filters against the API schema.

Parameters:

• endpoint (str): API endpoint (e.g., “/vn”, “/character”)

• filters (Union[List, str, None]): Filter expression to validate

Returns:

• Dict[str, Any]: Validation result

async VNDB.get_available_fields(endpoint: str) → List[str]

Get all available filterable fields for an endpoint.

Get available fields for an endpoint.

Parameters:

• endpoint (str): API endpoint

Returns:

• List[str]: Available field names

async VNDB.list_endpoints() → List[str]

Get all available API endpoints.

Get all available API endpoints.

Returns:

• List[str]: Available endpoints

Cache Management

VNDB.invalidate_schema_cache()

Invalidates the schema cache, forcing a refresh on next validation or schema access.

Invalidate the schema cache, forcing a refresh on next use.

async VNDB.update_local_schema() → Dict[str, Any]

Forces a download of the latest schema and updates the local schema file (if configured) or the cache.

Force download of the latest schema.

Returns:

• Dict[str, Any]: Updated schema data

Context Manager Support

All clients support async context manager protocol:

async with VNDB() as client:
    # Client is automatically opened and closed
    response = await client.vn.query(query)
# Session is automatically closed here

This ensures proper cleanup of HTTP sessions and other resources.