Skip to content

API.md

Latest commit

 

History

History
263 lines (185 loc) · 4.67 KB

API.md

File metadata and controls

263 lines (185 loc) · 4.67 KB
Raw

API Reference

Configuration

Settings

Application configuration from environment variables.

from ai_project.config import Settings, get_settings

settings = get_settings()
print(settings.app_env)  # "development"
print(settings.log_level)  # "INFO"

Attributes:

  • app_env: str - Application environment
  • app_debug: bool - Debug mode
  • log_level: str - Logging level
  • data_input_path: Path - Input directory
  • data_output_path: Path - Output directory
  • data_cache_enabled: bool - Enable caching
  • feature_scaling_method: str - Scaling method
  • feature_selection_threshold: float - Selection threshold

Logging

get_logger(name: str)

Get structured logger instance.

from ai_project.logger import get_logger

logger = get_logger(__name__)
logger.info("Processing started", file="data.csv")

configure_logging(level: str)

Configure logging system.

from ai_project.logger import configure_logging

configure_logging("DEBUG")

Models

DataRecord

Single data record with validation.

from ai_project.models import DataRecord

record = DataRecord(
    id=1,
    value=10.5,
    label="test",
    timestamp=datetime.now()
)

Validators:

  • value >= 0 - Non-negative values
  • label - Non-empty string

ProcessingResult

Result of data processing operation.

from ai_project.models import ProcessingResult

result = ProcessingResult(
    success=True,
    records_processed=100,
    records_failed=0,
    errors=[],
    metadata={"duration": 1.5}
)

Processors

DataLoader

Load and save data files.

from ai_project.processors.loader import DataLoader

# Load CSV
df = DataLoader.load_csv("data.csv")

# Load JSON
df = DataLoader.load_json("data.json")

# Save CSV
DataLoader.save_csv(df, "output.csv")

DataValidator

Validate data against schema.

from ai_project.processors.validator import DataValidator

# Validate single record
record = DataValidator.validate_record({"id": 1, "value": 10.0, "label": "test"})

# Validate DataFrame
valid_records, errors = DataValidator.validate_dataframe(df)

# Validate schema
DataValidator.validate_schema(df, ["id", "value", "label"])

DataCleaner

Clean and preprocess data.

from ai_project.processors.cleaner import DataCleaner

# Remove duplicates
df = DataCleaner.remove_duplicates(df)

# Remove nulls
df = DataCleaner.remove_nulls(df)

# Normalize column
df = DataCleaner.normalize_numeric(df, "value", method="minmax")

# Full cleaning pipeline
df = DataCleaner.clean(df)

FeatureEngineer

Create and engineer features.

from ai_project.processors.feature_eng import FeatureEngineer

# Scale features
df = FeatureEngineer.scale_features(df, ["value"], method="standard")

# Select features
df_selected, cols = FeatureEngineer.select_features(df, threshold=0.8)

# Create interaction
df = FeatureEngineer.create_interaction_features(df, "id", "value")

Pipeline

DataPipeline

Orchestrate complete data pipeline.

from ai_project.core.pipeline import DataPipeline
from ai_project.core.models import PipelineConfig

config = PipelineConfig(
    name="my_pipeline",
    validate_data=True,
    clean_data=True,
    engineer_features=True,
    scaling_method="standard"
)

pipeline = DataPipeline(config)
result = pipeline.execute("input.csv")

print(result.success)  # True
print(result.records_processed)  # 100
print(result.duration_seconds)  # 1.23

PipelineConfig

Configuration for pipeline execution.

from ai_project.core.models import PipelineConfig

config = PipelineConfig(
    name="default",
    validate_data=True,
    clean_data=True,
    engineer_features=False,
    scaling_method="standard"
)

PipelineResult

Result of pipeline execution.

from ai_project.core.models import PipelineResult

result = PipelineResult(
    success=True,
    pipeline_name="my_pipeline",
    records_processed=100,
    records_failed=0,
    duration_seconds=1.23,
    errors=[],
    metadata={"output_rows": 100}
)

Exceptions

Exception Hierarchy

from ai_project.exceptions import (
    AIProjectError,
    ConfigurationError,
    DataLoadError,
    ValidationError,
    ProcessingError,
    PipelineError,
)

try:
    df = DataLoader.load_csv("missing.csv")
except DataLoadError as e:
    print(f"Failed to load: {e}")

CLI

Commands

# Process data
ai-project process input.csv --output output.csv

# Validate data
ai-project validate input.csv

# Show info
ai-project info

Options

# Enable debug mode
ai-project process input.csv --debug

# Custom output
ai-project process input.csv --output custom.csv