Query Loader API

The query_loader module manages Cypher query files for cleaner code organization.

Overview

Instead of embedding Cypher queries in Python code, the library stores queries in .cypher files within the cypher_queries directory. The query loader provides a simple interface to load and cache these queries.

Classes

QueryLoader

Class-based interface for loading Cypher queries with caching support.

from ifc_graph import QueryLoader

# Create loader
loader = QueryLoader(use_cache=True)

# List available queries
queries = loader.list_queries()
print(queries)  # ['clear_database', 'create_elements_batch', ...]

# Load a query
query = loader.load("create_elements_batch")
print(query)

Class-based interface for loading Cypher queries.

Provides an alternative to the module-level functions for object-oriented usage patterns.

Source code in src/ifc_graph/query_loader.py

class QueryLoader:
    """
    Class-based interface for loading Cypher queries.

    Provides an alternative to the module-level functions for
    object-oriented usage patterns.
    """

    def __init__(self, queries_dir: Optional[Path] = None, use_cache: bool = True):
        """
        Initialize the query loader.

        Args:
            queries_dir: Optional custom queries directory
            use_cache: Whether to cache loaded queries
        """
        self._queries_dir = queries_dir or get_queries_directory()
        self._use_cache = use_cache
        self._cache: dict[str, str] = {}

    @property
    def queries_dir(self) -> Path:
        """Get the queries directory."""
        return self._queries_dir

    def load(self, query_name: str) -> str:
        """
        Load a query by name.

        Args:
            query_name: Name of the query file (without .cypher extension)

        Returns:
            The Cypher query string
        """
        if self._use_cache and query_name in self._cache:
            return self._cache[query_name]

        query_path = self._queries_dir / f"{query_name}.cypher"

        if not query_path.exists():
            raise FileNotFoundError(f"Query file not found: {query_path}")

        query = query_path.read_text(encoding='utf-8')

        if self._use_cache:
            self._cache[query_name] = query

        return query

    def list_queries(self) -> list[str]:
        """List all available query names."""
        if not self._queries_dir.exists():
            return []
        return [f.stem for f in self._queries_dir.glob("*.cypher")]

    def clear_cache(self) -> None:
        """Clear the query cache."""
        self._cache.clear()

queries_dir property

Get the queries directory.

__init__(queries_dir=None, use_cache=True)

Initialize the query loader.

Parameters:

Name	Type	Description	Default
queries_dir	Optional[Path]	Optional custom queries directory	None
use_cache	bool	Whether to cache loaded queries	True

Source code in src/ifc_graph/query_loader.py

def __init__(self, queries_dir: Optional[Path] = None, use_cache: bool = True):
    """
    Initialize the query loader.

    Args:
        queries_dir: Optional custom queries directory
        use_cache: Whether to cache loaded queries
    """
    self._queries_dir = queries_dir or get_queries_directory()
    self._use_cache = use_cache
    self._cache: dict[str, str] = {}

clear_cache()

Clear the query cache.

Source code in src/ifc_graph/query_loader.py

def clear_cache(self) -> None:
    """Clear the query cache."""
    self._cache.clear()

list_queries()

List all available query names.

Source code in src/ifc_graph/query_loader.py

def list_queries(self) -> list[str]:
    """List all available query names."""
    if not self._queries_dir.exists():
        return []
    return [f.stem for f in self._queries_dir.glob("*.cypher")]

load(query_name)

Load a query by name.

Parameters:

Name	Type	Description	Default
query_name	str	Name of the query file (without .cypher extension)	required

Returns:

Type	Description
str	The Cypher query string

Source code in src/ifc_graph/query_loader.py

def load(self, query_name: str) -> str:
    """
    Load a query by name.

    Args:
        query_name: Name of the query file (without .cypher extension)

    Returns:
        The Cypher query string
    """
    if self._use_cache and query_name in self._cache:
        return self._cache[query_name]

    query_path = self._queries_dir / f"{query_name}.cypher"

    if not query_path.exists():
        raise FileNotFoundError(f"Query file not found: {query_path}")

    query = query_path.read_text(encoding='utf-8')

    if self._use_cache:
        self._cache[query_name] = query

    return query

Constructor Parameters

Parameter	Type	Default	Description
queries_dir	Path	None	Custom queries directory (default: package directory)
use_cache	bool	True	Whether to cache loaded queries

Methods

load(query_name)

Load a query by name.

loader = QueryLoader()
query = loader.load("create_elements_batch")

Parameters:

Parameter	Type	Description
query_name	str	Name of query file (without .cypher extension)

Returns: The Cypher query string

Raises: FileNotFoundError if query file doesn't exist

list_queries()

List all available query names.

loader = QueryLoader()
available = loader.list_queries()
# ['clear_database', 'create_elements_batch', 'create_materials_batch', ...]

Returns: List of query names (without .cypher extension)

clear_cache()

Clear the query cache.

loader = QueryLoader(use_cache=True)
loader.load("create_elements_batch")  # Loads from file
loader.load("create_elements_batch")  # Uses cache
loader.clear_cache()
loader.load("create_elements_batch")  # Loads from file again

Properties

queries_dir

Get the queries directory path.

loader = QueryLoader()
print(loader.queries_dir)
# /path/to/site-packages/ifc_graph/cypher_queries

Functions

load_query

Module-level function to load a Cypher query.

from ifc_graph import load_query

query = load_query("create_elements_batch")
print(query)

Load a Cypher query from a .cypher file.

Parameters:

Name	Type	Description	Default
query_name	str	Name of the query file (without .cypher extension)	required
use_cache	bool	Whether to use cached queries (default: True)	True

Returns:

Type	Description
str	The Cypher query string

Raises:

Type	Description
FileNotFoundError	If the query file doesn't exist
IOError	If there's an error reading the file

Source code in src/ifc_graph/query_loader.py

def load_query(query_name: str, use_cache: bool = True) -> str:
    """
    Load a Cypher query from a .cypher file.

    Args:
        query_name: Name of the query file (without .cypher extension)
        use_cache: Whether to use cached queries (default: True)

    Returns:
        The Cypher query string

    Raises:
        FileNotFoundError: If the query file doesn't exist
        IOError: If there's an error reading the file
    """
    if use_cache and query_name in _query_cache:
        logger.debug(f"Using cached query: {query_name}")
        return _query_cache[query_name]

    query_path = get_queries_directory() / f"{query_name}.cypher"

    if not query_path.exists():
        raise FileNotFoundError(f"Query file not found: {query_path}")

    try:
        query = query_path.read_text(encoding='utf-8')

        # Cache the query
        if use_cache:
            _query_cache[query_name] = query

        logger.debug(f"Loaded query from file: {query_name}")
        return query

    except IOError as e:
        logger.error(f"Error reading query file {query_path}: {e}")
        raise

Parameters

Parameter	Type	Default	Description
query_name	str	-	Name of query file (without .cypher extension)
use_cache	bool	True	Whether to use cached queries

Returns

The Cypher query string.

Raises

• FileNotFoundError: If query file doesn't exist
• IOError: If there's an error reading the file

list_available_queries

List all available query files.

from ifc_graph.query_loader import list_available_queries

queries = list_available_queries()
print(queries)

List all available query files.

Returns:

Type	Description
list[str]	List of query names (without .cypher extension)

Source code in src/ifc_graph/query_loader.py

def list_available_queries() -> list[str]:
    """
    List all available query files.

    Returns:
        List of query names (without .cypher extension)
    """
    queries_dir = get_queries_directory()

    if not queries_dir.exists():
        return []

    return [f.stem for f in queries_dir.glob("*.cypher")]

validate_queries

Validate that all query files exist and are readable.

from ifc_graph.query_loader import validate_queries

results = validate_queries()
for name, valid in results.items():
    status = "OK" if valid else "FAILED"
    print(f"{name}: {status}")

Validate that all query files exist and are readable.

Returns:

Type	Description
dict[str, bool]	Dictionary mapping query names to their validity status

Source code in src/ifc_graph/query_loader.py

def validate_queries() -> dict[str, bool]:
    """
    Validate that all query files exist and are readable.

    Returns:
        Dictionary mapping query names to their validity status
    """
    results = {}

    for query_name in list_available_queries():
        try:
            load_query(query_name, use_cache=False)
            results[query_name] = True
        except Exception as e:
            logger.warning(f"Invalid query {query_name}: {e}")
            results[query_name] = False

    return results

clear_cache

Clear the module-level query cache.

from ifc_graph.query_loader import clear_cache

clear_cache()

Clear the query cache.

Source code in src/ifc_graph/query_loader.py

def clear_cache() -> None:
    """Clear the query cache."""
    _query_cache.clear()
    logger.debug("Query cache cleared")

get_queries_directory

Get the path to the cypher_queries directory.

from ifc_graph.query_loader import get_queries_directory

queries_dir = get_queries_directory()
print(queries_dir)

Get the path to the cypher_queries directory within the package.

Source code in src/ifc_graph/query_loader.py

def get_queries_directory() -> Path:
    """Get the path to the cypher_queries directory within the package."""
    return Path(__file__).parent / "cypher_queries"

Available Queries

The package includes these pre-defined queries:

Query Name	Description
clear_database	Delete all nodes and relationships
create_project	Create the IFC project node
create_elements_batch	Batch create element nodes
create_structures_batch	Batch create structure nodes
create_materials_batch	Batch create materials and relationships
create_project_contains	Link project to elements
create_structure_contains	Link structures to elements
create_relationships_batch	Create element relationships
create_property_sets_batch	Batch create property sets
create_metadata	Create import metadata node

Query File Format

Query files are plain text files with the .cypher extension containing valid Cypher queries.

Example: create_elements_batch.cypher

// Batch create Element nodes from a list of elements
// Parameters: $elements - list of element objects with properties
UNWIND $elements AS elem
CREATE (e:Element {
    id: elem.id,
    name: elem.name,
    guid: elem.guid,
    type: elem.type,
    object_type: elem.object_type,
    description: elem.description,
    tag: elem.tag
})
RETURN count(e) AS created_count

Custom Query Directory

You can use a custom directory for your own queries:

from pathlib import Path
from ifc_graph import QueryLoader

# Use custom queries directory
my_queries = Path("/path/to/my/queries")
loader = QueryLoader(queries_dir=my_queries)

# Load custom query
query = loader.load("my_custom_query")

Example: Using Queries with Neo4j

from ifc_graph import load_query, Neo4jConnection

# Load a query
query = load_query("create_elements_batch")

# Execute with Neo4j
with Neo4jConnection(uri, user, password) as conn:
    with conn.session() as session:
        elements = [
            {'id': '1', 'name': 'Wall 1', 'guid': 'abc', 'type': 'IfcWall', ...},
            {'id': '2', 'name': 'Wall 2', 'guid': 'def', 'type': 'IfcWall', ...},
        ]
        result = session.run(query, elements=elements)
        count = result.single()['created_count']
        print(f"Created {count} elements")

Caching Behavior

The query loader uses caching to avoid reading files repeatedly:

from ifc_graph import load_query

# First call reads from file
query1 = load_query("create_elements_batch", use_cache=True)

# Second call uses cache (faster)
query2 = load_query("create_elements_batch", use_cache=True)

# Force file read
query3 = load_query("create_elements_batch", use_cache=False)

To clear the cache:

from ifc_graph.query_loader import clear_cache

clear_cache()

See Also

• Element Filter API - IFC extraction
• Neo4j Store API - Database operations
• Graph Model - Database schema