Neo4j Store API

The neo4j_store module handles Neo4j database operations including connection management and batch data insertion.

Classes

Neo4jConnection

Manages Neo4j database connections with retry logic and proper cleanup.

from ifc_graph import Neo4jConnection

# Using context manager (recommended)
with Neo4jConnection(
    uri="bolt://localhost:7687",
    username="neo4j",
    password="password",
    max_retries=3,
    retry_delay=1.0
) as conn:
    with conn.session() as session:
        result = session.run("MATCH (n) RETURN count(n)")
        print(result.single()[0])

Manages Neo4j database connections with retry logic and proper cleanup.

Source code in src/ifc_graph/neo4j_store.py

class Neo4jConnection:
    """
    Manages Neo4j database connections with retry logic and proper cleanup.
    """

    def __init__(
        self,
        uri: str,
        username: str,
        password: str,
        max_retries: int = 3,
        retry_delay: float = 1.0
    ):
        self.uri = uri
        self.username = username
        self.password = password
        self.max_retries = max_retries
        self.retry_delay = retry_delay
        self._driver = None

    def connect(self) -> None:
        """
        Establish connection to Neo4j database with retry logic.

        Raises:
            DatabaseConnectionError: If connection cannot be established
        """
        last_error = None

        for attempt in range(1, self.max_retries + 1):
            try:
                logger.info(f"Connecting to Neo4j (attempt {attempt}/{self.max_retries})...")
                self._driver = GraphDatabase.driver(
                    self.uri,
                    auth=(self.username, self.password)
                )
                # Verify connection
                self._driver.verify_connectivity()
                logger.info("Successfully connected to Neo4j")
                return

            except AuthError as e:
                raise DatabaseConnectionError(
                    f"Authentication failed. Check username/password: {e}"
                ) from e

            except ServiceUnavailable as e:
                last_error = e
                if attempt < self.max_retries:
                    logger.warning(
                        f"Connection attempt {attempt} failed. Retrying in {self.retry_delay}s..."
                    )
                    time.sleep(self.retry_delay)

            except Exception as e:
                raise DatabaseConnectionError(
                    f"Unexpected error connecting to Neo4j: {e}"
                ) from e

        raise DatabaseConnectionError(
            f"Failed to connect after {self.max_retries} attempts: {last_error}"
        )

    def close(self) -> None:
        """Close the database connection."""
        if self._driver:
            self._driver.close()
            self._driver = None
            logger.info("Neo4j connection closed")

    @contextmanager
    def session(self):
        """Context manager for database sessions."""
        if not self._driver:
            raise DatabaseConnectionError("Not connected to database")

        session = self._driver.session()
        try:
            yield session
        finally:
            session.close()

    def __enter__(self):
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

__init__(uri, username, password, max_retries=3, retry_delay=1.0)

Source code in src/ifc_graph/neo4j_store.py

def __init__(
    self,
    uri: str,
    username: str,
    password: str,
    max_retries: int = 3,
    retry_delay: float = 1.0
):
    self.uri = uri
    self.username = username
    self.password = password
    self.max_retries = max_retries
    self.retry_delay = retry_delay
    self._driver = None

connect()

Establish connection to Neo4j database with retry logic.

Raises:

Type	Description
DatabaseConnectionError	If connection cannot be established

Source code in src/ifc_graph/neo4j_store.py

def connect(self) -> None:
    """
    Establish connection to Neo4j database with retry logic.

    Raises:
        DatabaseConnectionError: If connection cannot be established
    """
    last_error = None

    for attempt in range(1, self.max_retries + 1):
        try:
            logger.info(f"Connecting to Neo4j (attempt {attempt}/{self.max_retries})...")
            self._driver = GraphDatabase.driver(
                self.uri,
                auth=(self.username, self.password)
            )
            # Verify connection
            self._driver.verify_connectivity()
            logger.info("Successfully connected to Neo4j")
            return

        except AuthError as e:
            raise DatabaseConnectionError(
                f"Authentication failed. Check username/password: {e}"
            ) from e

        except ServiceUnavailable as e:
            last_error = e
            if attempt < self.max_retries:
                logger.warning(
                    f"Connection attempt {attempt} failed. Retrying in {self.retry_delay}s..."
                )
                time.sleep(self.retry_delay)

        except Exception as e:
            raise DatabaseConnectionError(
                f"Unexpected error connecting to Neo4j: {e}"
            ) from e

    raise DatabaseConnectionError(
        f"Failed to connect after {self.max_retries} attempts: {last_error}"
    )

close()

Close the database connection.

Source code in src/ifc_graph/neo4j_store.py

def close(self) -> None:
    """Close the database connection."""
    if self._driver:
        self._driver.close()
        self._driver = None
        logger.info("Neo4j connection closed")

session()

Context manager for database sessions.

Source code in src/ifc_graph/neo4j_store.py

@contextmanager
def session(self):
    """Context manager for database sessions."""
    if not self._driver:
        raise DatabaseConnectionError("Not connected to database")

    session = self._driver.session()
    try:
        yield session
    finally:
        session.close()

Constructor Parameters

Parameter	Type	Default	Description
uri	str	-	Neo4j connection URI (e.g., "bolt://localhost:7687")
username	str	-	Neo4j username
password	str	-	Neo4j password
max_retries	int	3	Maximum connection retry attempts
retry_delay	float	1.0	Delay between retries in seconds

Methods

connect()

Establish connection to Neo4j with retry logic.

conn = Neo4jConnection(uri, username, password)
conn.connect()  # Raises DatabaseConnectionError on failure

close()

Close the database connection.

conn.close()

session()

Context manager for database sessions.

with conn.session() as session:
    session.run("MATCH (n) RETURN n")

Context Manager Usage

# Automatically connects and disconnects
with Neo4jConnection(uri, user, password) as conn:
    with conn.session() as session:
        # Use session
        pass
# Connection closed automatically

Exceptions

DatabaseConnectionError

Raised when there's an error connecting to the database.

from ifc_graph import DatabaseConnectionError

try:
    with Neo4jConnection(uri, user, password) as conn:
        pass
except DatabaseConnectionError as e:
    print(f"Connection failed: {e}")

Common causes: - Neo4j not running - Wrong URI - Invalid credentials - Network issues

Bases: Exception

Raised when there's an error connecting to the database.

Source code in src/ifc_graph/neo4j_store.py

class DatabaseConnectionError(Exception):
    """Raised when there's an error connecting to the database."""
    pass

DatabaseOperationError

Raised when a database operation fails.

from ifc_graph import DatabaseOperationError

try:
    stats = save_to_neo4j(elements, ifc_file, uri, user, password)
except DatabaseOperationError as e:
    print(f"Operation failed: {e}")

Common causes: - Query syntax error - Constraint violation - Transaction timeout

Bases: Exception

Raised when a database operation fails.

Source code in src/ifc_graph/neo4j_store.py

class DatabaseOperationError(Exception):
    """Raised when a database operation fails."""
    pass

Functions

save_to_neo4j

The main function for saving IFC elements to Neo4j.

from ifc_graph import filter_physical_elements, save_to_neo4j

elements, ifc_file = filter_physical_elements("model.ifc")

stats = save_to_neo4j(
    filtered_elements=elements,
    ifc_file=ifc_file,
    uri="bolt://localhost:7687",
    username="neo4j",
    password="password",
    clear_db=False,
    config={
        'include_materials': True,
        'include_property_sets': True,
        'max_properties_per_element': 50,
    }
)

print(f"Elements: {stats['elements']}")
print(f"Structures: {stats['structures']}")
print(f"Materials: {stats['materials']}")

Saves filtered IFC elements to Neo4j database using batch operations.

Parameters:

Name	Type	Description	Default
filtered_elements	dict	Dictionary of element type -> elements list	required
ifc_file	Any	Loaded IFC file object	required
uri	str	Neo4j connection URI	required
username	str	Neo4j username	required
password	str	Neo4j password	required
clear_db	bool	Whether to clear the database before import (default: False)	False
config	Optional[dict]	Extraction configuration	None

Returns:

Type	Description
dict	Dictionary with import statistics

Raises:

Type	Description
DatabaseConnectionError	If connection fails
DatabaseOperationError	If database operations fail

Source code in src/ifc_graph/neo4j_store.py

def save_to_neo4j(
    filtered_elements: dict,
    ifc_file: Any,
    uri: str,
    username: str,
    password: str,
    clear_db: bool = False,
    config: Optional[dict] = None
) -> dict:
    """
    Saves filtered IFC elements to Neo4j database using batch operations.

    Args:
        filtered_elements: Dictionary of element type -> elements list
        ifc_file: Loaded IFC file object
        uri: Neo4j connection URI
        username: Neo4j username
        password: Neo4j password
        clear_db: Whether to clear the database before import (default: False)
        config: Extraction configuration

    Returns:
        Dictionary with import statistics

    Raises:
        DatabaseConnectionError: If connection fails
        DatabaseOperationError: If database operations fail
    """
    if config is None:
        config = {
            'include_materials': True,
            'include_property_sets': True,
            'max_properties_per_element': 50,
        }

    start_time = time.time()
    stats = {
        'elements': 0,
        'structures': 0,
        'materials': 0,
        'property_sets': 0,
    }

    try:
        with Neo4jConnection(uri, username, password) as conn:
            with conn.session() as session:
                # Optionally clear database
                if clear_db:
                    clear_database(session)

                # Create project node
                project = ifc_file.by_type("IfcProject")[0]
                project_id = create_project_node(session, project)

                # Process all elements
                all_spatial_info = []
                all_materials = []
                all_element_ids = []

                for element_type, elements in filtered_elements.items():
                    logger.info(f"Processing {len(elements)} {element_type} elements...")

                    count, spatial, materials, psets = batch_create_elements(
                        session, elements, config
                    )

                    stats['elements'] += count
                    all_spatial_info.extend(spatial)
                    all_materials.extend(materials)

                    # Collect element IDs for project relationships
                    for elem in elements:
                        all_element_ids.append(str(elem.id()))

                # Create relationships
                batch_create_project_relationships(session, project_id, all_element_ids)

                # Create structures and their relationships
                stats['structures'] = batch_create_structures(session, all_spatial_info)
                batch_create_structure_relationships(session, all_spatial_info)

                # Create materials
                stats['materials'] = batch_create_materials(session, all_materials)

                # Create metadata
                duration = time.time() - start_time
                create_metadata(session, {
                    'timestamp': time.strftime("%Y-%m-%d %H:%M:%S"),
                    'element_count': stats['elements'],
                    'types': ", ".join(filtered_elements.keys()),
                    'ifc_file': str(ifc_file.header.file_name.name if hasattr(ifc_file, 'header') else 'Unknown'),
                    'duration': round(duration, 2),
                })

        save_time = time.time() - start_time
        logger.info(f"Saved {stats['elements']} elements to Neo4j in {save_time:.2f} seconds")

        return stats

    except DatabaseConnectionError:
        raise
    except Neo4jError as e:
        raise DatabaseOperationError(f"Database operation failed: {e}") from e
    except Exception as e:
        raise DatabaseOperationError(f"Unexpected error during save: {e}") from e

Parameters

Parameter	Type	Default	Description
filtered_elements	dict	-	Dictionary from filter_physical_elements()
ifc_file	ifcopenshell.file	-	Loaded IFC file object
uri	str	-	Neo4j connection URI
username	str	-	Neo4j username
password	str	-	Neo4j password
clear_db	bool	False	Clear database before import
config	dict	None	Extraction configuration

Return Value

Returns a dictionary with import statistics:

{
    'elements': 150,      # Element nodes created
    'structures': 12,     # Structure nodes created
    'materials': 45,      # Material relationships created
    'property_sets': 0,   # Reserved for future use
}

Internal Functions

These functions are used internally by save_to_neo4j() but can be used directly for custom workflows.

clear_database

Clear all nodes and relationships from the database.

from ifc_graph.neo4j_store import clear_database

with Neo4jConnection(uri, user, password) as conn:
    with conn.session() as session:
        clear_database(session)

Warning

This will delete all data in the database.

create_project_node

Create the IFC project node.

from ifc_graph.neo4j_store import create_project_node

with conn.session() as session:
    project = ifc_file.by_type("IfcProject")[0]
    project_id = create_project_node(session, project)

batch_create_elements

Create element nodes in batches for performance.

from ifc_graph.neo4j_store import batch_create_elements

with conn.session() as session:
    count, spatial, materials, psets = batch_create_elements(
        session,
        elements=wall_elements,
        config={'include_materials': True},
        batch_size=500
    )

batch_create_structures

Create structure nodes (storeys, buildings, etc.).

from ifc_graph.neo4j_store import batch_create_structures

with conn.session() as session:
    count = batch_create_structures(session, spatial_info, batch_size=500)

batch_create_materials

Create material nodes and HAS_MATERIAL relationships.

from ifc_graph.neo4j_store import batch_create_materials

with conn.session() as session:
    count = batch_create_materials(session, materials, batch_size=500)

create_metadata

Create import metadata node.

from ifc_graph.neo4j_store import create_metadata

with conn.session() as session:
    create_metadata(session, {
        'timestamp': '2024-01-15 10:30:00',
        'element_count': 150,
        'types': 'IfcWall, IfcDoor, IfcWindow',
        'ifc_file': 'building.ifc',
        'duration': 2.5,
    })

Configuration Options

The config parameter accepts:

Option	Type	Default	Description
include_materials	bool	True	Create material nodes and relationships
include_property_sets	bool	True	Extract and store property sets
max_properties_per_element	int	50	Maximum properties per element

Example: Custom Database Operations

from ifc_graph import (
    Neo4jConnection,
    filter_physical_elements,
)
from ifc_graph.neo4j_store import (
    clear_database,
    create_project_node,
    batch_create_elements,
    batch_create_structures,
    batch_create_materials,
)

# Load IFC
elements, ifc_file = filter_physical_elements("model.ifc")

# Custom database operations
with Neo4jConnection(uri, user, password) as conn:
    with conn.session() as session:
        # Clear existing data
        clear_database(session)

        # Create project
        project = ifc_file.by_type("IfcProject")[0]
        project_id = create_project_node(session, project)

        # Process elements type by type
        for elem_type, elems in elements.items():
            print(f"Processing {len(elems)} {elem_type}...")

            count, spatial, materials, psets = batch_create_elements(
                session, elems, config={}, batch_size=100
            )

            batch_create_structures(session, spatial)
            batch_create_materials(session, materials)

        print("Done!")

See Also

• Element Filter API - IFC extraction
• Query Loader API - Cypher query management
• Graph Model - Database schema