API Reference¶
Python API reference for grai.build.
Core Models¶
Core Pydantic models for grai.build.
This module defines the data structures for entities, relations, and properties that form the foundation of the declarative knowledge graph modeling.
Entity
¶
Bases: BaseModel
Represents a node/entity in the knowledge graph.
Attributes:
Name | Type | Description |
---|---|---|
entity |
str
|
The entity type name (becomes the node label in Neo4j). |
source |
Union[str, SourceConfig]
|
The data source - can be a string or SourceConfig object. |
keys |
List[str]
|
List of property names that uniquely identify this entity. |
properties |
List[Property]
|
List of properties/attributes for this entity. |
description |
Optional[str]
|
Optional description of the entity. |
metadata |
Dict[str, Any]
|
Optional additional metadata. |
Source code in grai/core/models.py
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 |
|
get_key_properties()
¶
Get all properties that are designated as keys.
Returns:
Type | Description |
---|---|
List[Property]
|
List of Property objects that are keys. |
get_property(name)
¶
Get a property by name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name
|
str
|
The property name to look up. |
required |
Returns:
Type | Description |
---|---|
Optional[Property]
|
The Property object if found, None otherwise. |
Source code in grai/core/models.py
get_source_config()
¶
Get the full source configuration.
Returns:
Type | Description |
---|---|
SourceConfig
|
SourceConfig object. |
Source code in grai/core/models.py
get_source_name()
¶
Get the source name as a string.
Returns:
Type | Description |
---|---|
str
|
Source name string. |
validate_entity_name(v)
classmethod
¶
Validate that entity name is a valid identifier.
Source code in grai/core/models.py
validate_keys(v)
classmethod
¶
Validate that all keys are non-empty.
validate_source(v)
classmethod
¶
Convert string source to SourceConfig for consistency.
Source code in grai/core/models.py
Project
¶
Bases: BaseModel
Represents a complete grai.build project configuration.
Attributes:
Name | Type | Description |
---|---|---|
name |
str
|
The project name. |
version |
str
|
The project version. |
entities |
List[Entity]
|
List of entity definitions in the project. |
relations |
List[Relation]
|
List of relation definitions in the project. |
config |
Dict[str, Any]
|
Optional project-level configuration. |
Source code in grai/core/models.py
get_entity(name)
¶
Get an entity by name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name
|
str
|
The entity name to look up. |
required |
Returns:
Type | Description |
---|---|
Optional[Entity]
|
The Entity object if found, None otherwise. |
Source code in grai/core/models.py
Property
¶
Bases: BaseModel
Represents a property (attribute) of an entity or relation.
Attributes:
Name | Type | Description |
---|---|---|
name |
str
|
The property name. |
type |
PropertyType
|
The data type of the property. |
required |
bool
|
Whether this property must have a value. |
description |
Optional[str]
|
Optional description of the property. |
default |
Optional[Any]
|
Optional default value for the property. |
Source code in grai/core/models.py
validate_name(v)
classmethod
¶
Validate that property name is a valid identifier.
Source code in grai/core/models.py
PropertyType
¶
Bases: str
, Enum
Supported property types for entity and relation attributes.
Source code in grai/core/models.py
Relation
¶
Bases: BaseModel
Represents an edge/relation in the knowledge graph.
Attributes:
Name | Type | Description |
---|---|---|
relation |
str
|
The relation type name (becomes the edge label in Neo4j). |
from_entity |
str
|
The source entity type. |
to_entity |
str
|
The target entity type. |
source |
Union[str, SourceConfig]
|
The data source - can be a string or SourceConfig object. |
mappings |
RelationMapping
|
How source and target entities connect via keys. |
properties |
List[Property]
|
List of properties/attributes for this relation. |
description |
Optional[str]
|
Optional description of the relation. |
metadata |
Dict[str, Any]
|
Optional additional metadata. |
Source code in grai/core/models.py
223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 |
|
get_property(name)
¶
Get a property by name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name
|
str
|
The property name to look up. |
required |
Returns:
Type | Description |
---|---|
Optional[Property]
|
The Property object if found, None otherwise. |
Source code in grai/core/models.py
get_source_config()
¶
Get the full source configuration.
Returns:
Type | Description |
---|---|
SourceConfig
|
SourceConfig object. |
Source code in grai/core/models.py
get_source_name()
¶
Get the source name as a string.
Returns:
Type | Description |
---|---|
str
|
Source name string. |
validate_relation_name(v)
classmethod
¶
Validate that relation name is uppercase and valid.
Source code in grai/core/models.py
validate_source(v)
classmethod
¶
Convert string source to SourceConfig for consistency.
Source code in grai/core/models.py
RelationMapping
¶
Bases: BaseModel
Defines how entities are connected in a relation.
Attributes:
Name | Type | Description |
---|---|---|
from_key |
str
|
The key property name on the source entity. |
to_key |
str
|
The key property name on the target entity. |
Source code in grai/core/models.py
SourceConfig
¶
Bases: BaseModel
Configuration for entity/relation data sources.
Supports both simple string format (backward compatible) and detailed config.
Attributes:
Name | Type | Description |
---|---|---|
name |
str
|
Source identifier (e.g., table name, file path, API endpoint). |
type |
Optional[SourceType]
|
Type of data source. |
connection |
Optional[str]
|
Optional connection string or identifier. |
schema |
Optional[str]
|
Optional database schema name. |
database |
Optional[str]
|
Optional database name. |
format |
Optional[str]
|
Optional data format details. |
metadata |
Dict[str, Any]
|
Optional additional source metadata. |
Source code in grai/core/models.py
from_string(source)
classmethod
¶
Create a SourceConfig from a simple string.
Maintains backward compatibility with existing entity definitions.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
source
|
str
|
Simple source string (e.g., "analytics.customers") |
required |
Returns:
Type | Description |
---|---|
SourceConfig
|
SourceConfig with inferred type if possible. |
Source code in grai/core/models.py
SourceType
¶
Bases: str
, Enum
Supported source types for entities and relations.
Source code in grai/core/models.py
Entity¶
from grai.core.models import Entity, Property
entity = Entity(
entity="customer",
source="analytics.customers",
keys=["customer_id"],
properties=[
Property(name="customer_id", type="string"),
Property(name="email", type="string"),
]
)
Relation¶
from grai.core.models import Relation, RelationMappings
relation = Relation(
relation="PURCHASED",
from_entity="customer",
to_entity="product",
source="analytics.orders",
mappings=RelationMappings(
from_key="customer_id",
to_key="product_id"
)
)
Parser¶
Parser module for loading YAML definitions into Pydantic models.
ParserError
¶
Bases: Exception
Base exception for parser errors.
Source code in grai/core/parser/yaml_parser.py
__init__(message, file_path=None)
¶
Initialize parser error.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
message
|
str
|
Error message. |
required |
file_path
|
Optional[Path]
|
Optional path to the file that caused the error. |
None
|
Source code in grai/core/parser/yaml_parser.py
ValidationParserError
¶
Bases: ParserError
Exception raised when Pydantic validation fails.
Source code in grai/core/parser/yaml_parser.py
YAMLParseError
¶
Bases: ParserError
Exception raised when YAML parsing fails.
Source code in grai/core/parser/yaml_parser.py
load_entities_from_directory(directory)
¶
Load all entity definitions from a directory.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
directory
|
Union[str, Path]
|
Path to directory containing entity YAML files. |
required |
Returns:
Type | Description |
---|---|
List[Entity]
|
List of Entity instances. |
Raises:
Type | Description |
---|---|
ParserError
|
If parsing any file fails. |
Source code in grai/core/parser/yaml_parser.py
load_project(project_root, entities_dir='entities', relations_dir='relations', manifest_file='grai.yml')
¶
Load a complete grai.build project from a directory structure.
Expected structure
project_root/ ├── grai.yml ├── entities/ │ ├── entity1.yml │ └── entity2.yml └── relations/ └── relation1.yml
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project_root
|
Union[str, Path]
|
Root directory of the project. |
required |
entities_dir
|
str
|
Subdirectory containing entity definitions (default: "entities"). |
'entities'
|
relations_dir
|
str
|
Subdirectory containing relation definitions (default: "relations"). |
'relations'
|
manifest_file
|
str
|
Name of the project manifest file (default: "grai.yml"). |
'grai.yml'
|
Returns:
Type | Description |
---|---|
Project
|
Project instance with all entities and relations loaded. |
Raises:
Type | Description |
---|---|
ParserError
|
If loading fails. |
Source code in grai/core/parser/yaml_parser.py
load_project_manifest(file_path='grai.yml')
¶
Load the project manifest (grai.yml).
Parameters:
Name | Type | Description | Default |
---|---|---|---|
file_path
|
Union[str, Path]
|
Path to the grai.yml file (default: "grai.yml"). |
'grai.yml'
|
Returns:
Type | Description |
---|---|
Dict[str, Any]
|
Dictionary containing project configuration. |
Raises:
Type | Description |
---|---|
ParserError
|
If the file cannot be loaded. |
Source code in grai/core/parser/yaml_parser.py
load_relations_from_directory(directory)
¶
Load all relation definitions from a directory.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
directory
|
Union[str, Path]
|
Path to directory containing relation YAML files. |
required |
Returns:
Type | Description |
---|---|
List[Relation]
|
List of Relation instances. |
Raises:
Type | Description |
---|---|
ParserError
|
If parsing any file fails. |
Source code in grai/core/parser/yaml_parser.py
parse_entity_file(file_path)
¶
Parse an entity definition from a YAML file.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
file_path
|
Union[str, Path]
|
Path to the entity YAML file. |
required |
Returns:
Type | Description |
---|---|
Entity
|
Entity instance. |
Raises:
Type | Description |
---|---|
ParserError
|
If parsing fails. |
Source code in grai/core/parser/yaml_parser.py
parse_relation_file(file_path)
¶
Parse a relation definition from a YAML file.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
file_path
|
Union[str, Path]
|
Path to the relation YAML file. |
required |
Returns:
Type | Description |
---|---|
Relation
|
Relation instance. |
Raises:
Type | Description |
---|---|
ParserError
|
If parsing fails. |
Source code in grai/core/parser/yaml_parser.py
YAML Parser¶
from grai.core.parser.yaml_parser import YAMLParser
parser = YAMLParser()
# Parse entity file
entity = parser.parse_entity_file("entities/customer.yml")
# Parse relation file
relation = parser.parse_relation_file("relations/purchased.yml")
# Parse entire project
project = parser.parse_project(".")
Validator¶
Validator module for checking project consistency and correctness.
EntityReferenceError
¶
Bases: ValidationError
Exception raised when an entity reference is invalid.
Source code in grai/core/validator/validator.py
KeyMappingError
¶
Bases: ValidationError
Exception raised when a key mapping is invalid.
Source code in grai/core/validator/validator.py
ValidationError
¶
Bases: Exception
Base exception for validation errors.
Source code in grai/core/validator/validator.py
__init__(message, context=None)
¶
Initialize validation error.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
message
|
str
|
Error message. |
required |
context
|
Optional[str]
|
Optional context (e.g., entity or relation name). |
None
|
Source code in grai/core/validator/validator.py
ValidationResult
¶
Result of a validation operation.
Attributes:
Name | Type | Description |
---|---|---|
valid |
bool
|
Whether validation passed. |
errors |
List[str]
|
List of validation errors. |
warnings |
List[str]
|
List of validation warnings. |
Source code in grai/core/validator/validator.py
__bool__()
¶
__init__()
¶
__str__()
¶
Return string representation of validation result.
Source code in grai/core/validator/validator.py
add_error(message, context=None)
¶
Add an error to the result.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
message
|
str
|
Error message. |
required |
context
|
Optional[str]
|
Optional context. |
None
|
Source code in grai/core/validator/validator.py
add_warning(message, context=None)
¶
Add a warning to the result.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
message
|
str
|
Warning message. |
required |
context
|
Optional[str]
|
Optional context. |
None
|
Source code in grai/core/validator/validator.py
validate_entity(entity)
¶
Validate a single entity.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
entity
|
Entity
|
The entity to validate. |
required |
Returns:
Type | Description |
---|---|
ValidationResult
|
ValidationResult with any errors or warnings. |
Source code in grai/core/validator/validator.py
validate_entity_references(relations, entity_index, result=None)
¶
Validate that all entity references in relations exist.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
relations
|
List[Relation]
|
List of relations to validate. |
required |
entity_index
|
Dict[str, Entity]
|
Index of entities by name. |
required |
result
|
Optional[ValidationResult]
|
Optional existing ValidationResult to add to. |
None
|
Returns:
Type | Description |
---|---|
ValidationResult
|
ValidationResult with any errors found. |
Source code in grai/core/validator/validator.py
validate_key_mappings(relations, entity_index, result=None)
¶
Validate that key mappings in relations reference valid entity keys.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
relations
|
List[Relation]
|
List of relations to validate. |
required |
entity_index
|
Dict[str, Entity]
|
Index of entities by name. |
required |
result
|
Optional[ValidationResult]
|
Optional existing ValidationResult to add to. |
None
|
Returns:
Type | Description |
---|---|
ValidationResult
|
ValidationResult with any errors found. |
Source code in grai/core/validator/validator.py
validate_project(project, strict=True)
¶
Validate an entire project for consistency and correctness.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
The project to validate. |
required |
strict
|
bool
|
If True, warnings will be treated as errors. |
True
|
Returns:
Type | Description |
---|---|
ValidationResult
|
ValidationResult with all errors and warnings. |
Raises:
Type | Description |
---|---|
ValidationError
|
If strict=True and validation fails. |
Source code in grai/core/validator/validator.py
validate_relation(relation, entity_index=None)
¶
Validate a single relation.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
relation
|
Relation
|
The relation to validate. |
required |
entity_index
|
Optional[Dict[str, Entity]]
|
Optional index of entities for reference checking. |
None
|
Returns:
Type | Description |
---|---|
ValidationResult
|
ValidationResult with any errors or warnings. |
Source code in grai/core/validator/validator.py
Schema Validator¶
from grai.core.validator.validator import Validator
from grai.core.models import Project
validator = Validator()
project = Project(...)
# Validate project
result = validator.validate(project)
if result.is_valid:
print("✅ Validation passed")
else:
for error in result.errors:
print(f"❌ {error}")
Compiler¶
Compiler module for generating database queries from models.
CompilerError
¶
compile_and_write(project, output_dir='target/neo4j', filename='compiled.cypher', include_header=True, include_constraints=True)
¶
Compile a project and write the Cypher script to a file.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
Project to compile. |
required |
output_dir
|
Union[str, Path]
|
Directory to write the output file. |
'target/neo4j'
|
filename
|
str
|
Name of the output file. |
'compiled.cypher'
|
include_header
|
bool
|
If True, include script header. |
True
|
include_constraints
|
bool
|
If True, include constraint statements. |
True
|
Returns:
Type | Description |
---|---|
Path
|
Path to the written file. |
Raises:
Type | Description |
---|---|
CompilerError
|
If compilation or writing fails. |
Source code in grai/core/compiler/cypher_compiler.py
compile_entity(entity)
¶
Compile an entity into a Cypher MERGE statement.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
entity
|
Entity
|
Entity model to compile. |
required |
Returns:
Type | Description |
---|---|
str
|
Cypher MERGE statement for creating/updating nodes. |
Example
Source code in grai/core/compiler/cypher_compiler.py
compile_project(project, include_header=True, include_constraints=True)
¶
Compile a complete project into a Cypher script.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
Project model to compile. |
required |
include_header
|
bool
|
If True, include script header with project info. |
True
|
include_constraints
|
bool
|
If True, include constraint creation statements. |
True
|
Returns:
Type | Description |
---|---|
str
|
Complete Cypher script as a string. |
Source code in grai/core/compiler/cypher_compiler.py
167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 |
|
compile_relation(relation)
¶
Compile a relation into Cypher MATCH...MERGE statements.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
relation
|
Relation
|
Relation model to compile. |
required |
Returns:
Type | Description |
---|---|
str
|
Cypher statements for creating relationships. |
Example
Source code in grai/core/compiler/cypher_compiler.py
compile_schema_only(project)
¶
Compile only the schema (constraints and indexes) without data loading.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
Project to compile schema for. |
required |
Returns:
Type | Description |
---|---|
str
|
Cypher script with only schema definitions. |
Source code in grai/core/compiler/cypher_compiler.py
generate_load_csv_statements(project, data_dir='data')
¶
Generate LOAD CSV statements for entities and relations.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
Project to generate load statements for. |
required |
data_dir
|
str
|
Directory containing CSV files. |
'data'
|
Returns:
Type | Description |
---|---|
Dict[str, str]
|
Dictionary mapping entity/relation names to LOAD CSV statements. |
Source code in grai/core/compiler/cypher_compiler.py
write_cypher_file(cypher, output_path, create_dirs=True)
¶
Write Cypher script to a file.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
cypher
|
str
|
Cypher script content. |
required |
output_path
|
Union[str, Path]
|
Path to write the file. |
required |
create_dirs
|
bool
|
If True, create parent directories if they don't exist. |
True
|
Returns:
Type | Description |
---|---|
Path
|
Path to the written file. |
Raises:
Type | Description |
---|---|
CompilerError
|
If file cannot be written. |
Source code in grai/core/compiler/cypher_compiler.py
Cypher Compiler¶
from grai.core.compiler.cypher_compiler import CypherCompiler
from grai.core.models import Entity, Relation
compiler = CypherCompiler()
# Compile entity
entity_cypher = compiler.compile_entity(entity)
# Compile relation
relation_cypher = compiler.compile_relation(relation)
# Compile entire project
project_cypher = compiler.compile_project(project)
Loader¶
Loader module for executing Cypher against Neo4j and loading data from warehouses.
Neo4jConnection
dataclass
¶
Neo4j connection configuration.
Attributes:
Name | Type | Description |
---|---|---|
uri |
str
|
Neo4j connection URI (e.g., bolt://localhost:7687) |
user |
str
|
Username for authentication |
password |
str
|
Password for authentication |
database |
str
|
Database name (default: neo4j) |
encrypted |
bool
|
Whether to use encrypted connection |
max_retry_time |
int
|
Maximum time to retry connection (seconds) |
Source code in grai/core/loader/neo4j_loader.py
check_apoc_available(driver, database='neo4j')
¶
Check if APOC library is installed and available in Neo4j.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
database
|
str
|
Database name to check. |
'neo4j'
|
Returns:
Type | Description |
---|---|
bool
|
True if APOC is available, False otherwise. |
Example
Source code in grai/core/loader/neo4j_loader.py
check_indexes_exist(driver, label, properties, database='neo4j')
¶
Check if indexes exist for specified label and properties.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
label
|
str
|
Node label to check. |
required |
properties
|
List[str]
|
List of property names to check for indexes. |
required |
database
|
str
|
Database name to check. |
'neo4j'
|
Returns:
Type | Description |
---|---|
Dict[str, bool]
|
Dictionary mapping property names to whether an index exists. |
Example
Source code in grai/core/loader/neo4j_loader.py
close_connection(driver)
¶
Close Neo4j driver connection.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance to close. |
required |
Source code in grai/core/loader/neo4j_loader.py
connect_neo4j(connection_or_uri=None, user=None, password=None, database='neo4j', encrypted=False, max_retry_time=30, uri=None)
¶
Connect to Neo4j database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
connection_or_uri
|
Optional[Union[Neo4jConnection, str]]
|
Either a Neo4jConnection object or a URI string |
None
|
user
|
Optional[str]
|
Username for authentication (required if connection_or_uri is a string) |
None
|
password
|
Optional[str]
|
Password for authentication (required if connection_or_uri is a string) |
None
|
database
|
str
|
Database name (default: neo4j) |
'neo4j'
|
encrypted
|
bool
|
Whether to use encrypted connection |
False
|
max_retry_time
|
int
|
Maximum time to retry connection (seconds) |
30
|
uri
|
Optional[str]
|
(Legacy) URI string - use connection_or_uri instead |
None
|
Returns:
Type | Description |
---|---|
Driver
|
Neo4j driver instance. |
Raises:
Type | Description |
---|---|
ImportError
|
If neo4j driver is not installed. |
ServiceUnavailable
|
If cannot connect to Neo4j. |
AuthError
|
If authentication fails. |
Example
# Option 1: Using Neo4jConnection object
conn = Neo4jConnection(uri="bolt://localhost:7687", user="neo4j", password="password")
driver = connect_neo4j(conn)
# Option 2: Using individual parameters
driver = connect_neo4j(
uri="bolt://localhost:7687",
user="neo4j",
password="password"
)
# Option 3: Positional URI
driver = connect_neo4j(
"bolt://localhost:7687",
user="neo4j",
password="password"
)
Source code in grai/core/loader/neo4j_loader.py
98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
|
execute_apoc_periodic_iterate(driver, batch_data, cypher_statement, batch_size=10000, parallel=False, database='neo4j')
¶
Execute APOC periodic.iterate for efficient bulk loading.
This uses APOC's apoc.periodic.iterate which automatically handles: - Sub-batch commits (reduces memory usage) - Parallel processing (optional) - Better performance for large datasets
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
batch_data
|
List[Dict[str, Any]]
|
List of dictionaries to load. |
required |
cypher_statement
|
str
|
Cypher statement to execute for each row. Use 'row' to reference each item. |
required |
batch_size
|
int
|
Number of rows to process per sub-batch. |
10000
|
parallel
|
bool
|
Whether to use parallel processing. |
False
|
database
|
str
|
Database name. |
'neo4j'
|
Returns:
Type | Description |
---|---|
ExecutionResult
|
ExecutionResult with execution details. |
Example
data = [
{"customer_id": "C001", "name": "Alice"},
{"customer_id": "C002", "name": "Bob"}
]
cypher = '''
MERGE (c:Customer {customer_id: row.customer_id})
SET c.name = row.name
'''
result = execute_apoc_periodic_iterate(
driver, data, cypher, batch_size=10000
)
print(f"Loaded {result.records_affected} rows")
Source code in grai/core/loader/neo4j_loader.py
459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 |
|
execute_cypher(driver, cypher, parameters=None, database='neo4j')
¶
Execute Cypher statement(s) against Neo4j.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
cypher
|
str
|
Cypher statement(s) to execute. |
required |
parameters
|
Optional[Dict[str, Any]]
|
Optional parameters for the query. |
None
|
database
|
str
|
Database name to execute against. |
'neo4j'
|
Returns:
Type | Description |
---|---|
ExecutionResult
|
ExecutionResult with execution details. |
Example
Source code in grai/core/loader/neo4j_loader.py
574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 |
|
execute_cypher_file(driver, file_path, database='neo4j', batch_size=None)
¶
Execute Cypher statements from a file.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
file_path
|
Union[str, Path]
|
Path to Cypher file. |
required |
database
|
str
|
Database name to execute against. |
'neo4j'
|
batch_size
|
Optional[int]
|
Optional batch size for large files. |
None
|
Returns:
Type | Description |
---|---|
ExecutionResult
|
ExecutionResult with execution details. |
Raises:
Type | Description |
---|---|
FileNotFoundError
|
If file does not exist. |
Example
Source code in grai/core/loader/neo4j_loader.py
get_apoc_version(driver, database='neo4j')
¶
Get the installed APOC version.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
database
|
str
|
Database name. |
'neo4j'
|
Returns:
Type | Description |
---|---|
Optional[str]
|
APOC version string or None if not available. |
Source code in grai/core/loader/neo4j_loader.py
get_database_info(driver, database='neo4j')
¶
Get information about the Neo4j database.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
database
|
str
|
Database name. |
'neo4j'
|
Returns:
Type | Description |
---|---|
Dict[str, Any]
|
Dictionary with database information. |
Example
Source code in grai/core/loader/neo4j_loader.py
verify_connection(driver, database='neo4j')
¶
Verify that connection to Neo4j is working.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
database
|
str
|
Database name to test. |
'neo4j'
|
Returns:
Type | Description |
---|---|
bool
|
True if connection is working, False otherwise. |
Source code in grai/core/loader/neo4j_loader.py
verify_indexes_and_warn(driver, label, key_properties, database='neo4j', verbose=True)
¶
Verify that indexes exist for key properties and print warnings if missing.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
driver
|
Driver
|
Neo4j driver instance. |
required |
label
|
str
|
Node label to check. |
required |
key_properties
|
List[str]
|
List of key property names that should have indexes. |
required |
database
|
str
|
Database name to check. |
'neo4j'
|
verbose
|
bool
|
Whether to print warnings. |
True
|
Returns:
Type | Description |
---|---|
bool
|
True if all indexes exist, False otherwise. |
Example
Source code in grai/core/loader/neo4j_loader.py
Neo4j Loader¶
from grai.core.loader.neo4j_loader import (
connect_neo4j,
execute_cypher,
close_connection,
)
# Connect
driver = connect_neo4j(
uri="bolt://localhost:7687",
user="neo4j",
password="password",
database="neo4j"
)
# Execute Cypher
cypher = "CREATE CONSTRAINT ..."
result = execute_cypher(driver, cypher)
if result.success:
print(f"✅ Executed {result.statements_executed} statements")
print(f" Nodes created: {result.nodes_created}")
print(f" Properties set: {result.properties_set}")
else:
for error in result.errors:
print(f"❌ {error}")
# Close
close_connection(driver)
BigQuery Loader¶
from grai.core.loader.bigquery_loader import (
BigQueryExtractor,
load_entity_from_bigquery,
)
# Extract data
extractor = BigQueryExtractor(
project_id="my-project",
credentials_path="service-account.json"
)
# Load entity
result = load_entity_from_bigquery(
entity=entity,
bigquery_connection=extractor,
neo4j_connection=driver,
batch_size=1000,
limit=None,
verbose=True
)
print(f"✅ Loaded {result.rows_processed} rows")
print(f" Duration: {result.duration_seconds}s")
Profiles¶
Profile management for grai.build connections.
Inspired by dbt's profiles.yml, this module handles connection configurations for data warehouses (BigQuery, PostgreSQL, Snowflake, etc.) and graph databases (Neo4j).
BigQueryProfile
¶
Bases: BaseModel
BigQuery connection profile.
Source code in grai/core/profiles.py
validate_method(v)
classmethod
¶
Validate authentication method.
Source code in grai/core/profiles.py
Neo4jProfile
¶
Bases: BaseModel
Neo4j connection profile.
Source code in grai/core/profiles.py
PostgresProfile
¶
Bases: BaseModel
PostgreSQL connection profile.
Source code in grai/core/profiles.py
validate_sslmode(v)
classmethod
¶
Validate SSL mode.
Source code in grai/core/profiles.py
Profile
¶
SnowflakeProfile
¶
Bases: BaseModel
Snowflake connection profile.
Source code in grai/core/profiles.py
TargetConfig
¶
Bases: BaseModel
Configuration for a specific target (environment).
Source code in grai/core/profiles.py
create_default_profiles_file()
¶
Create a default profiles.yml file.
Returns:
Type | Description |
---|---|
Path
|
Path to created profiles file |
Source code in grai/core/profiles.py
322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 |
|
get_connection_info(profile_name='default', target_name=None)
¶
Get warehouse and graph connection info from profiles.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
profile_name
|
str
|
Name of the profile to use (defaults to 'default') |
'default'
|
target_name
|
Optional[str]
|
Name of the target (defaults to profile's default or GRAI_TARGET) |
None
|
Returns:
Type | Description |
---|---|
tuple[Any, Neo4jProfile]
|
Tuple of (warehouse_profile, graph_profile) |
Raises:
Type | Description |
---|---|
FileNotFoundError
|
If profiles.yml doesn't exist |
KeyError
|
If profile or target doesn't exist |
Source code in grai/core/profiles.py
get_profile(profile_name)
¶
Get a specific profile by name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
profile_name
|
str
|
Name of the profile to retrieve |
required |
Returns:
Type | Description |
---|---|
Dict[str, Any]
|
Profile configuration dictionary |
Raises:
Type | Description |
---|---|
KeyError
|
If profile doesn't exist |
Source code in grai/core/profiles.py
get_profile_path()
¶
Get the path to profiles.yml.
Returns:
Type | Description |
---|---|
Path
|
Path to profiles.yml file |
get_profiles_dir()
¶
Get the profiles directory path.
Checks in order: 1. GRAI_PROFILES_DIR environment variable 2. ~/.grai/ directory
Returns:
Type | Description |
---|---|
Path
|
Path to profiles directory |
Source code in grai/core/profiles.py
get_target_config(profile_name, target_name=None)
¶
Get target configuration from a profile.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
profile_name
|
str
|
Name of the profile |
required |
target_name
|
Optional[str]
|
Name of the target (defaults to profile's default target) |
None
|
Returns:
Type | Description |
---|---|
Dict[str, Any]
|
Target configuration dictionary with 'warehouse' and 'graph' outputs |
Raises:
Type | Description |
---|---|
KeyError
|
If profile or target doesn't exist |
Source code in grai/core/profiles.py
load_profiles()
¶
Load profiles from profiles.yml.
Returns:
Type | Description |
---|---|
Dict[str, Any]
|
Dictionary of profile configurations |
Raises:
Type | Description |
---|---|
FileNotFoundError
|
If profiles.yml doesn't exist |
YAMLError
|
If profiles.yml is invalid |
Source code in grai/core/profiles.py
parse_graph_profile(config)
¶
Parse graph database configuration into profile model.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
config
|
Dict[str, Any]
|
Graph database configuration dictionary |
required |
Returns:
Type | Description |
---|---|
Neo4jProfile
|
Neo4jProfile model |
Raises:
Type | Description |
---|---|
ValueError
|
If graph type is unsupported |
Source code in grai/core/profiles.py
parse_warehouse_profile(config)
¶
Parse warehouse configuration into appropriate profile model.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
config
|
Dict[str, Any]
|
Warehouse configuration dictionary |
required |
Returns:
Type | Description |
---|---|
Any
|
Profile model (BigQueryProfile, SnowflakeProfile, PostgresProfile, etc.) |
Raises:
Type | Description |
---|---|
ValueError
|
If warehouse type is unsupported |
Source code in grai/core/profiles.py
resolve_env_vars(config)
¶
Resolve environment variable references in configuration.
Replaces strings like "{{ env_var('MY_VAR') }}" with environment variable values.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
config
|
Dict[str, Any]
|
Configuration dictionary |
required |
Returns:
Type | Description |
---|---|
Dict[str, Any]
|
Configuration with environment variables resolved |
Source code in grai/core/profiles.py
Profile Configuration¶
from grai.core.profiles import (
BigQueryProfile,
Neo4jProfile,
TargetConfig,
)
# BigQuery profile
bq_profile = BigQueryProfile(
project_id="my-project",
dataset="my_dataset",
credentials_path="/path/to/credentials.json"
)
# Neo4j profile
neo4j_profile = Neo4jProfile(
uri="bolt://localhost:7687",
user="neo4j",
password="password",
database="neo4j"
)
# Target config
target = TargetConfig(
bigquery=bq_profile,
neo4j=neo4j_profile
)
Lineage¶
Lineage tracking module for knowledge graph analysis.
Exports lineage tracking functions for analyzing entity relationships, dependencies, and impact analysis.
LineageEdge
dataclass
¶
Represents an edge in the lineage graph.
Attributes:
Name | Type | Description |
---|---|---|
from_node |
str
|
Source node ID |
to_node |
str
|
Target node ID |
relation_type |
str
|
Type of relationship (e.g., "depends_on", "produces") |
metadata |
Dict
|
Additional metadata about the edge |
Source code in grai/core/lineage/lineage_tracker.py
LineageGraph
dataclass
¶
Represents the complete lineage graph.
Attributes:
Name | Type | Description |
---|---|---|
nodes |
Dict[str, LineageNode]
|
Dictionary mapping node IDs to LineageNode objects |
edges |
List[LineageEdge]
|
List of LineageEdge objects |
entity_map |
Dict[str, str]
|
Mapping of entity names to node IDs |
relation_map |
Dict[str, str]
|
Mapping of relation names to node IDs |
source_map |
Dict[str, str]
|
Mapping of source names to node IDs |
Source code in grai/core/lineage/lineage_tracker.py
add_edge(edge)
¶
add_node(node)
¶
Add a node to the graph.
Source code in grai/core/lineage/lineage_tracker.py
get_edges_from(node_id)
¶
get_edges_to(node_id)
¶
LineageNode
dataclass
¶
Represents a node in the lineage graph.
Attributes:
Name | Type | Description |
---|---|---|
id |
str
|
Unique identifier for the node |
name |
str
|
Node name (entity name, relation name, or source) |
type |
NodeType
|
Type of node (entity, relation, or source) |
metadata |
Dict
|
Additional metadata about the node |
Source code in grai/core/lineage/lineage_tracker.py
NodeType
¶
build_lineage_graph(project)
¶
Build a complete lineage graph from a project.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
Project to analyze |
required |
Returns:
Type | Description |
---|---|
LineageGraph
|
LineageGraph with all entities, relations, and sources |
Source code in grai/core/lineage/lineage_tracker.py
112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 |
|
calculate_impact_analysis(graph, entity_name)
¶
Calculate the impact of changes to an entity.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
entity_name
|
str
|
Name of the entity to analyze |
required |
Returns:
Type | Description |
---|---|
Dict
|
Dictionary with impact analysis |
Source code in grai/core/lineage/lineage_tracker.py
export_lineage_to_dict(graph)
¶
Export lineage graph to dictionary format.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
Returns:
Type | Description |
---|---|
Dict
|
Dictionary representation of the graph |
Source code in grai/core/lineage/lineage_tracker.py
find_downstream_entities(graph, entity_name, max_depth=10)
¶
Find all downstream entities (recursive).
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
entity_name
|
str
|
Name of the entity |
required |
max_depth
|
int
|
Maximum depth to traverse |
10
|
Returns:
Type | Description |
---|---|
Set[str]
|
Set of downstream entity names |
Source code in grai/core/lineage/lineage_tracker.py
find_entity_path(graph, from_entity, to_entity)
¶
Find shortest path between two entities.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
from_entity
|
str
|
Starting entity name |
required |
to_entity
|
str
|
Target entity name |
required |
Returns:
Type | Description |
---|---|
Optional[List[str]]
|
List of node names representing the path, or None if no path exists |
Source code in grai/core/lineage/lineage_tracker.py
find_upstream_entities(graph, entity_name, max_depth=10)
¶
Find all upstream entities (recursive).
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
entity_name
|
str
|
Name of the entity |
required |
max_depth
|
int
|
Maximum depth to traverse |
10
|
Returns:
Type | Description |
---|---|
Set[str]
|
Set of upstream entity names |
Source code in grai/core/lineage/lineage_tracker.py
get_entity_lineage(graph, entity_name)
¶
Get complete lineage information for an entity.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
entity_name
|
str
|
Name of the entity |
required |
Returns:
Type | Description |
---|---|
Dict
|
Dictionary with lineage information |
Source code in grai/core/lineage/lineage_tracker.py
get_lineage_statistics(graph)
¶
Get statistics about the lineage graph.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
Returns:
Type | Description |
---|---|
Dict
|
Dictionary with statistics |
Source code in grai/core/lineage/lineage_tracker.py
get_relation_lineage(graph, relation_name)
¶
Get complete lineage information for a relation.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
relation_name
|
str
|
Name of the relation |
required |
Returns:
Type | Description |
---|---|
Dict
|
Dictionary with lineage information |
Source code in grai/core/lineage/lineage_tracker.py
visualize_lineage_graphviz(graph, focus_entity=None)
¶
Generate Graphviz DOT representation of lineage.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
focus_entity
|
Optional[str]
|
Optional entity to focus on (shows only related nodes) |
None
|
Returns:
Type | Description |
---|---|
str
|
Graphviz DOT diagram as string |
Source code in grai/core/lineage/lineage_tracker.py
visualize_lineage_mermaid(graph, focus_entity=None)
¶
Generate Mermaid diagram representation of lineage.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
graph
|
LineageGraph
|
Lineage graph |
required |
focus_entity
|
Optional[str]
|
Optional entity to focus on (shows only related nodes) |
None
|
Returns:
Type | Description |
---|---|
str
|
Mermaid diagram as string |
Source code in grai/core/lineage/lineage_tracker.py
Lineage Tracker¶
from grai.core.lineage.lineage_tracker import LineageTracker
tracker = LineageTracker()
# Build lineage graph
lineage = tracker.build_lineage(project)
# Export as Mermaid
mermaid = tracker.export_mermaid(lineage)
# Export as DOT
dot = tracker.export_dot(lineage)
# Export as JSON
json_data = tracker.export_json(lineage)
Visualizer¶
Interactive visualization module for knowledge graphs.
Provides HTML-based interactive visualizations using D3.js and other web technologies.
generate_cytoscape_visualization(project, output_path, title=None, width=1200, height=800)
¶
Generate interactive Cytoscape.js visualization of the knowledge graph.
Creates an HTML file with an interactive graph using Cytoscape.js.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
The Project to visualize |
required |
output_path
|
Path
|
Path to save the HTML file |
required |
title
|
Optional[str]
|
Optional title for the visualization (defaults to project name) |
None
|
width
|
int
|
Width of the visualization canvas in pixels |
1200
|
height
|
int
|
Height of the visualization canvas in pixels |
800
|
Example
from grai.core.parser.yaml_parser import load_project project = load_project(Path(".")) generate_cytoscape_visualization(project, Path("graph.html"))
Source code in grai/core/visualizer/__init__.py
generate_d3_visualization(project, output_path, title=None, width=1200, height=800)
¶
Generate interactive D3.js visualization of the knowledge graph.
Creates an HTML file with an interactive force-directed graph using D3.js.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
project
|
Project
|
The Project to visualize |
required |
output_path
|
Path
|
Path to save the HTML file |
required |
title
|
Optional[str]
|
Optional title for the visualization (defaults to project name) |
None
|
width
|
int
|
Width of the visualization canvas in pixels |
1200
|
height
|
int
|
Height of the visualization canvas in pixels |
800
|
Example
from grai.core.parser.yaml_parser import load_project project = load_project(Path(".")) generate_d3_visualization(project, Path("graph.html"))
Source code in grai/core/visualizer/__init__.py
Graph Visualizer¶
from grai.core.visualizer.visualizer import Visualizer
visualizer = Visualizer()
# Generate D3.js visualization
html = visualizer.generate_d3(project, output="graph.html")
# Generate Cytoscape.js visualization
html = visualizer.generate_cytoscape(project, output="graph.html")
# Generate custom visualization
html = visualizer.generate_custom(
project,
template="custom_template.html",
output="graph.html"
)
CLI¶
Main CLI¶
from grai.cli.main import main_cli
# Programmatically invoke CLI
if __name__ == "__main__":
main_cli()
Type Definitions¶
Common Types¶
from typing import Optional, List, Dict, Any
# Entity types
EntityName = str
PropertyName = str
PropertyType = str
# Source types
SourceReference = Optional[str]
# Cypher types
CypherStatement = str
CypherStatements = List[CypherStatement]
# Result types
ValidationResult = Dict[str, Any]
CompilationResult = Dict[str, Any]
ExecutionResult = Dict[str, Any]
Exceptions¶
Custom Exceptions¶
from grai.core.exceptions import (
GraiError,
ValidationError,
CompilationError,
ConnectionError,
ExecutionError,
)
try:
result = validator.validate(project)
except ValidationError as e:
print(f"Validation failed: {e}")
except GraiError as e:
print(f"General error: {e}")
Utilities¶
Common Utilities¶
from grai.core.utils import (
load_yaml,
write_yaml,
ensure_dir,
hash_file,
)
# Load YAML
data = load_yaml("grai.yml")
# Write YAML
write_yaml(data, "output.yml")
# Ensure directory exists
ensure_dir("target/neo4j")
# Hash file for caching
hash_value = hash_file("entities/customer.yml")
Usage Examples¶
Complete Workflow¶
from pathlib import Path
from grai.core.parser.yaml_parser import YAMLParser
from grai.core.validator.validator import Validator
from grai.core.compiler.cypher_compiler import CypherCompiler
from grai.core.loader.neo4j_loader import connect_neo4j, execute_cypher
# 1. Parse project
parser = YAMLParser()
project = parser.parse_project(Path.cwd())
# 2. Validate
validator = Validator()
result = validator.validate(project)
if not result.is_valid:
for error in result.errors:
print(f"❌ {error}")
exit(1)
# 3. Compile
compiler = CypherCompiler()
cypher = compiler.compile_project(project)
# 4. Execute
driver = connect_neo4j(
uri="bolt://localhost:7687",
user="neo4j",
password="password"
)
result = execute_cypher(driver, cypher)
print(f"✅ Executed successfully")
print(f" Nodes created: {result.nodes_created}")
print(f" Relationships created: {result.relationships_created}")
Data Loading Workflow¶
from grai.core.loader.bigquery_loader import (
BigQueryExtractor,
load_entity_from_bigquery,
)
# Setup connections
extractor = BigQueryExtractor(
project_id="my-project",
credentials_path="credentials.json"
)
driver = connect_neo4j(
uri="bolt://localhost:7687",
user="neo4j",
password="password"
)
# Load each entity
for entity in project.entities:
print(f"Loading {entity.entity}...")
result = load_entity_from_bigquery(
entity=entity,
bigquery_connection=extractor,
neo4j_connection=driver,
batch_size=1000,
verbose=True
)
if result.success:
print(f"✅ Loaded {result.rows_processed} rows")
else:
print(f"❌ Failed: {result.errors}")
See Also¶
- Getting Started - Tutorial and examples
- Command Reference - CLI commands
- YAML Schema - Configuration reference