System Query Operations - Stash Tools & Scrapers

Operations for querying system status and statistics. Bases: StashClientProtocol Mixin for system query methods.

Functions

get_system_status

get_system_status() -> SystemStatus | None

Get the current Stash system status. Returns:

Type	Description
`SystemStatus \| None`	SystemStatus object containing system information and status
`SystemStatus \| None`	None if the query fails

Examples: Check system status:

status = await client.get_system_status()
if status:
    print(f"System status: {status.status}")
    print(f"Database path: {status.databasePath}")

check_system_ready

check_system_ready() -> None

Check if the Stash system is ready for processing. This method queries the system status and raises an exception if the system is not in OK status. It should be called before starting any processing operations. Raises:

Type	Description
`StashSystemNotReadyError`	If system status is SETUP or NEEDS_MIGRATION
`RuntimeError`	If system status cannot be determined

Examples: Validate system before processing:

try:
    await client.check_system_ready()
    # Safe to proceed with processing
    await client.metadata_scan()
except StashSystemNotReadyError as e:
    print(f"System not ready: {e}")
    # Handle migration or setup

stats

stats() -> StatsResultType

Get system statistics. Returns:

Type	Description
`StatsResultType`	StatsResultType object containing counts and metrics for all entity types

Examples: Get system statistics:

stats = await client.stats()
print(f"Total scenes: {stats.scene_count}")
print(f"Total performers: {stats.performer_count}")
print(f"Total O-count: {stats.total_o_count}")
print(f"Library size: {stats.scenes_size / (1024**3):.2f} GB")

logs

logs() -> list[LogEntry]

Get system logs. Returns:

Type	Description
`list[LogEntry]`	List of LogEntry objects containing log messages with timestamps and levels

Examples: Get recent logs:

logs = await client.logs()
for log in logs[-10:]:  # Last 10 entries
    print(f"[{log.time}] {log.level}: {log.message}")

Filter by level:

from stash_graphql_client.types import LogLevel

logs = await client.logs()
errors = [log for log in logs if log.level == LogLevel.ERROR]
print(f"Found {len(errors)} errors")

dlna_status

dlna_status() -> DLNAStatus

Get DLNA server status. Returns:

Type	Description
`DLNAStatus`	DLNAStatus object containing DLNA server information

Name	Type	Description	Default
`path`	`str \| Path \| None`	The directory path to list (string or Path object). If None, returns root directories. Note: Path is converted to string and sent to Stash server.	`None`
`locale`	`str`	Desired collation locale (e.g., ‘en-US’, ‘pt-BR’). Default is ‘en’	`'en'`

Type	Description
`Directory`	Directory object containing path information and subdirectories

sql_query

sql_query(
    sql: str, args: list[Any] | None = None
) -> SQLQueryResult

Execute a SQL query that returns rows.

This is a DANGEROUS operation that executes arbitrary SQL against the Stash database. Use with extreme caution. Incorrect queries can corrupt your database or expose sensitive information.

Parameters:

Name	Type	Description	Default
`sql`	`str`	The SQL query string to execute	required
`args`	`list[Any] \| None`	Optional list of query parameters to bind to the SQL statement	`None`

Returns:

Type	Description
`SQLQueryResult`	SQLQueryResult containing columns and rows from the query result

Raises:

Type	Description
`Exception`	If the query execution fails or SQL support is not available

Examples: Query database for scenes:

result = await client.sql_query(
    "SELECT id, title FROM scenes WHERE rating100 > ?",
    args=[80]
)
print(f"Columns: {result.columns}")
for row in result.rows:
    print(f"Scene ID: {row[0]}, Title: {row[1]}")

Count performers by gender:

result = await client.sql_query(
    "SELECT gender, COUNT(*) FROM performers GROUP BY gender"
)
for row in result.rows:
    print(f"{row[0]}: {row[1]} performers")

This feature requires Stash version that supports SQL queries. The exact version requirement should be documented in the Stash API.

sql_exec

sql_exec(
    sql: str, args: list[Any] | None = None
) -> SQLExecResult

Execute a SQL statement without returning rows (INSERT, UPDATE, DELETE).

This is a DANGEROUS operation that executes arbitrary SQL against the Stash database. Use with EXTREME caution. Incorrect statements can corrupt or destroy your database. Always backup before using this method.

Parameters:

Name	Type	Description	Default
`sql`	`str`	The SQL statement string to execute	required
`args`	`list[Any] \| None`	Optional list of statement parameters to bind to the SQL	`None`

Returns:

Type	Description
`SQLExecResult`	SQLExecResult containing rows_affected and last_insert_id

Raises:

Type	Description
`Exception`	If the statement execution fails or SQL support is not available

Examples: Update scene ratings:

result = await client.sql_exec(
    "UPDATE scenes SET rating100 = ? WHERE id = ?",
    args=[95, "123"]
)
print(f"Rows affected: {result.rows_affected}")

Delete orphaned tags (BE CAREFUL!):

result = await client.sql_exec(
    "DELETE FROM tags WHERE id NOT IN (SELECT tag_id FROM scene_tags)"
)
print(f"Deleted {result.rows_affected} orphaned tags")

Insert custom metadata:

result = await client.sql_exec(
    "INSERT INTO custom_metadata (entity_id, key, value) VALUES (?, ?, ?)",
    args=["scene-123", "custom_field", "custom_value"]
)
print(f"Last insert ID: {result.last_insert_id}")

This feature requires Stash version that supports SQL execution. The exact version requirement should be documented in the Stash API. ALWAYS test queries in a development environment before production use.

​Functions

​get_system_status

​check_system_ready

​stats

​logs

​dlna_status

​directory

​sql_query

​sql_exec

Functions

get_system_status

check_system_ready

stats

logs

dlna_status

directory

sql_query

sql_exec