tessl install tessl/pypi-kedro@1.1.0Kedro helps you build production-ready data and analytics pipelines
Agent Success
Agent success rate when using this tile
98%
Improvement
Agent success rate improvement when using this tile compared to baseline
1.32x
Baseline
Agent success rate without this tile
74%
Project initialization and metadata management.
class ProjectMetadata:
"""
Named tuple-like class containing project configuration and metadata.
Holds essential information about a Kedro project's structure and configuration.
"""
def __init__(
self,
config_file: Path,
package_name: str,
project_name: str,
project_path: Path,
source_dir: Path,
kedro_init_version: str,
tools: list,
example_pipeline: str
):
"""
Initialize ProjectMetadata.
Parameters:
- config_file: Path to pyproject.toml configuration file
- package_name: Python package name (import name)
- project_name: Project display name (human-readable)
- project_path: Project root path (contains conf/, src/, etc.)
- source_dir: Source code directory path (typically 'src')
- kedro_init_version: Version of Kedro used to initialize the project
- tools: List of additional tools/plugins configured (or None)
- example_pipeline: Name of example pipeline if configured (or None)
"""
@property
def source_dir(self) -> Path:
"""
Get source directory path.
Returns:
Path to source directory containing Python packages
"""
@property
def config_file(self) -> Path:
"""
Get configuration file path.
Returns:
Path to pyproject.toml or similar configuration file
"""
@property
def package_name(self) -> str:
"""
Get package name.
Returns:
Python package name used for imports
"""
@property
def project_name(self) -> str:
"""
Get project name.
Returns:
Human-readable project display name
"""
@property
def project_path(self) -> Path:
"""
Get project root path.
Returns:
Path to project root directory
"""
@property
def kedro_init_version(self) -> str:
"""
Get Kedro initialization version.
Returns:
Version of Kedro used to create the project
"""
@property
def tools(self) -> list:
"""
Get list of configured tools/plugins.
Returns:
List of tool names configured for the project, or None if not configured
"""
@property
def example_pipeline(self) -> str:
"""
Get example pipeline name.
Returns:
Name of example pipeline if configured, None otherwise
"""def bootstrap_project(project_path: Path) -> ProjectMetadata:
"""
Bootstrap a Kedro project by discovering and loading metadata.
Reads pyproject.toml to extract project information.
Parameters:
- project_path: Path to Kedro project root directory
Returns:
ProjectMetadata instance with project information
Raises:
RuntimeError: If project metadata cannot be found or parsed
Example:
>>> from pathlib import Path
>>> from kedro.framework.startup import bootstrap_project
>>> metadata = bootstrap_project(Path.cwd())
>>> print(metadata.package_name)
>>> print(metadata.kedro_init_version)
"""from pathlib import Path
from kedro.framework.startup import bootstrap_project
# Bootstrap project from current directory
metadata = bootstrap_project(Path.cwd())
# Access metadata
print(f"Project: {metadata.project_name}")
print(f"Package: {metadata.package_name}")
print(f"Kedro version: {metadata.kedro_init_version}")
print(f"Source dir: {metadata.source_dir}")
print(f"Config file: {metadata.config_file}")
print(f"Tools: {metadata.tools}")
print(f"Example pipeline: {metadata.example_pipeline}")Kedro CLI automatically bootstraps projects:
# In cli.py
from kedro.framework.startup import bootstrap_project
from pathlib import Path
@click.command()
def run():
"""Run the pipeline."""
# CLI automatically bootstraps the project
metadata = bootstrap_project(Path.cwd())
# Create session using metadata
with KedroSession.create(
project_path=metadata.project_path,
env="local"
) as session:
session.run()ProjectMetadata is passed to CLI hooks automatically:
from kedro.framework.cli.hooks import cli_hook_impl
class ProjectCLIHooks:
@cli_hook_impl
def before_command_run(self, project_metadata, command_args):
"""Execute before any CLI command runs."""
print(f"Running command in: {project_metadata.project_name}")
print(f"Package: {project_metadata.package_name}")
print(f"Project path: {project_metadata.project_path}")
# Validate project structure
if not (project_metadata.project_path / "conf").exists():
raise RuntimeError("Missing conf/ directory")
@cli_hook_impl
def after_command_run(self, project_metadata, command_args):
"""Execute after any CLI command runs."""
print(f"Command completed in: {project_metadata.project_name}")from kedro.framework.startup import bootstrap_project
from pathlib import Path
import sys
metadata = bootstrap_project(Path.cwd())
# Add source directory to Python path
sys.path.insert(0, str(metadata.source_dir))
# Dynamically import project modules
pipeline_module = f"{metadata.package_name}.pipeline_registry"
hooks_module = f"{metadata.package_name}.hooks"from kedro.framework.startup import bootstrap_project
from pathlib import Path
def validate_project_structure(project_path: Path) -> bool:
"""Validate Kedro project structure."""
try:
metadata = bootstrap_project(project_path)
# Check required directories
required_dirs = [
metadata.project_path / "conf",
metadata.project_path / "data",
metadata.source_dir / metadata.package_name
]
for directory in required_dirs:
if not directory.exists():
print(f"Missing directory: {directory}")
return False
# Check configuration file
if not metadata.config_file.exists():
print(f"Missing config file: {metadata.config_file}")
return False
print(f"✓ Project structure valid for: {metadata.project_name}")
return True
except RuntimeError as e:
print(f"Invalid project: {e}")
return FalseProjectMetadata flows through Kedro's components:
bootstrap_project()
↓
ProjectMetadata
↓
CLI Commands (via cli_hooks)
↓
KedroSession.create()
↓
KedroContext (via PACKAGE_NAME global)
↓
Pipeline Registry & HooksSee also: