StashContext

Context manager for managing the StashClient lifecycle.

StashContext

StashContext(
    conn: dict[str, Any] | None = None,
    verify_ssl: bool | str = True,
)

Context manager for Stash client.

This class provides a high-level interface for managing Stash client connections, including: - Connection configuration - Client lifecycle management - Interface access - Reference counting for safe concurrent usage

Example

# Create context with connection details
context = StashContext(conn={
    "Scheme": "http",
    "Host": "localhost",
    "Port": 9999,
    "ApiKey": "your_api_key",
})

# Use context manager (safe for concurrent tasks with singleton)
async with context as client:
    performer = await client.find_performer("123")

    # Access entity store for caching and advanced queries
    store = context.store
    scene = await store.get(Scene, "456")

# Or use directly
client = context.client
store = context.store
performer = await client.find_performer("123")
await context.close()

# Safe for concurrent tasks when using singleton
await asyncio.gather(
    task1(context),  # Each task uses async with
    task2(context),  # Reference counting prevents premature closure
)

Initialize context.

Parameters:

Name	Type	Description	Default
conn	dict[str, Any] | None	Connection details dictionary (case-insensitive keys)	None
verify_ssl	bool | str	Whether to verify SSL certificates (accepts bool or string like "true"/"false")	True

Source code in stash_graphql_client/context.py

def __init__(
    self,
    conn: dict[str, Any] | None = None,
    verify_ssl: bool | str = True,
) -> None:
    """Initialize context.

    Args:
        conn: Connection details dictionary (case-insensitive keys)
        verify_ssl: Whether to verify SSL certificates (accepts bool or string like "true"/"false")
    """
    # Normalize connection keys to canonical case
    self.conn = self._normalize_conn_keys(conn or {})
    # Convert string to bool if needed
    if isinstance(verify_ssl, str):
        self.verify_ssl = verify_ssl.lower() in ("true", "1", "yes")
    else:
        self.verify_ssl = verify_ssl
    self._client: StashClient | None = None
    self._store: StashEntityStore | None = None
    self._ref_count: int = 0
    self._ref_lock: asyncio.Lock = asyncio.Lock()

Attributes

conn instance-attribute

conn = _normalize_conn_keys(conn or {})

verify_ssl instance-attribute

verify_ssl = lower() in ('true', '1', 'yes')

interface property

interface: StashClient

Get Stash interface (alias for client).

Returns:

Type	Description
StashClient	StashClient instance

Raises:

Type	Description
RuntimeError	If client is not initialized

client property

client: StashClient

Get client instance.

Returns:

Type	Description
StashClient	StashClient instance

Raises:

Type	Description
RuntimeError	If client is not initialized

store property

store: StashEntityStore

Get entity store instance.

Returns:

Type	Description
StashEntityStore	StashEntityStore instance wired to StashObject identity map

Raises:

Type	Description
RuntimeError	If store is not initialized

capabilities property

capabilities: ServerCapabilities

Get detected server capabilities.

Returns:

Type	Description
ServerCapabilities	ServerCapabilities instance with feature flags for the connected server

Raises:

Type	Description
RuntimeError	If client is not initialized

ref_count property

ref_count: int

Get current reference count.

Returns:

Type	Description
int	Number of active async context managers using this context

Functions

get_client async

get_client() -> StashClient

Get initialized Stash client.

Returns:

Type	Description
StashClient	StashClient instance

Raises:

Type	Description
RuntimeError	If client initialization fails

Source code in stash_graphql_client/context.py

async def get_client(self) -> StashClient:
    """Get initialized Stash client.

    Returns:
        StashClient instance

    Raises:
        RuntimeError: If client initialization fails
    """
    logger.debug(
        f"get_client called on {id(self)}, current _client: {self._client}"
    )
    if self._client is None:
        # Normalize conn keys in case someone modified self.conn directly
        normalized_conn = self._normalize_conn_keys(self.conn)
        self._client = StashClient(
            conn=normalized_conn if normalized_conn else None,
            verify_ssl=self.verify_ssl,
        )
        try:
            await self._client.initialize()
            logger.debug(
                f"Client initialization complete, _client set to {self._client}"
            )

            # Initialize entity store and wire it to StashObject
            self._store = StashEntityStore(self._client)
            StashObject._store = self._store
            logger.debug("Entity store initialized and wired to StashObject")

        except Exception as e:
            logger.error(f"Client initialization failed: {e}")
            self._client = None
            raise RuntimeError(f"Failed to initialize Stash client: {e}")
    return self._client

close async

close(force: bool = False, _internal: bool = False) -> None

Close client connection.

Parameters:

Name	Type	Description	Default
force	bool	If True, force close even if reference count > 0. Use with caution as this may break concurrent tasks.	False
_internal	bool	Internal flag indicating call from aexit (already has lock)	False

Note

When using the singleton context with async context managers, manual close() is not needed - reference counting handles cleanup. Only call manually when you created the context directly and are done with it.

Source code in stash_graphql_client/context.py

async def close(self, force: bool = False, _internal: bool = False) -> None:
    """Close client connection.

    Args:
        force: If True, force close even if reference count > 0.
               Use with caution as this may break concurrent tasks.
        _internal: Internal flag indicating call from __aexit__ (already has lock)

    Note:
        When using the singleton context with async context managers,
        manual close() is not needed - reference counting handles cleanup.
        Only call manually when you created the context directly and are
        done with it.
    """
    # If called internally from __aexit__, we already have the lock
    if _internal:
        if self._client is not None:
            await self._client.close()
            self._client = None
            logger.debug("StashClient closed")
        if self._store is not None:
            # Clear store cache and unwire from StashObject
            self._store.invalidate_all()
            StashObject._store = None
            self._store = None
            logger.debug("Entity store cleared and unwired")
        return

    # Manual call - need to acquire lock
    async with self._ref_lock:
        if self._ref_count > 0 and not force:
            logger.warning(
                f"Attempted to close StashContext with {self._ref_count} active references. "
                "Close will be deferred until all context managers exit. "
                "Use force=True to override (not recommended)."
            )
            return

        if self._client is not None:
            await self._client.close()
            self._client = None
            logger.debug("StashClient closed")

        if self._store is not None:
            # Clear store cache and unwire from StashObject
            self._store.invalidate_all()
            StashObject._store = None
            self._store = None
            logger.debug("Entity store cleared and unwired")