coulomb/markitect-main
2
0
Fork 0
Code Issues 61 Pull Requests Actions Projects 12 Releases 1 Wiki Activity

feat(prompts): implement Phase 1 - Foundation (FR-1)

Browse Source 
Implement addressable artifacts with content-based identity and change detection.

Core Features:
- Artifact model with SHA-256 content digests
- ArtifactReference for cross-space addressing
- IArtifactRepository interface for pluggable storage
- SQLiteArtifactRepository implementation
- ArtifactService for high-level operations
- Content digest calculation utilities

Database:
- prompt_artifacts table with indexes
- Support for artifact metadata and types
- UNIQUE constraint on space_id+name

Tests (41 passing):
- 26 model tests (metadata, artifacts, references, digests)
- 15 repository tests (CRUD, queries, constraints)

Implements:
- FR-1.1: Unique addressability by name and ID
- FR-1.2: Content digest computation and storage
- FR-1.3: Cross-space artifact references

Files Created:
- markitect/prompts/models.py
- markitect/prompts/repositories/interfaces.py
- markitect/prompts/repositories/sqlite.py
- markitect/prompts/services/artifact_service.py
- migrations/prompts/001_create_artifacts_table.sql
- tests/unit/prompts/test_artifact_models.py
- tests/unit/prompts/test_artifact_repository.py

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2026-02-08 22:30:26 +01:00
parent cbde1dabc4
commit 945544880d
11 changed files with 1681 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
markitect/prompts/__init__.py Normal file
View File

@@ -0,0 +1,33 @@
"""
Prompt Dependency Resolution system for MarkiTect.
This package provides infrastructure for executing PromptTemplates with
deterministic dependency resolution, incremental recomputation, and quality
validation across InformationSpaces.
Key components:
- Artifact management with content-based addressing
- Template definition with macro support
- Deterministic resolution across spaces
- Idempotent execution with InputBundleHash
- Dependency graph construction and tracking
- Incremental recomputation with change impact analysis
- Quality gate validation and halting policies
- Complete traceability and provenance tracking
"""
__version__ = "0.1.0"
from markitect.prompts.models import (
Artifact,
ArtifactReference,
ArtifactMetadata,
ArtifactType,
)
__all__ = [
"Artifact",
"ArtifactReference",
"ArtifactMetadata",
"ArtifactType",
]

279
markitect/prompts/models.py Normal file
View File

@@ -0,0 +1,279 @@
"""
Core domain models for Prompt Dependency Resolution.
This module provides foundational data models for addressable artifacts,
references, and content hashing for change detection.
"""
import hashlib
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, Optional
from enum import Enum
class ArtifactType(Enum):
"""Type classification for artifacts."""
CONTENT = "content" # Regular content artifact
TEMPLATE = "template" # Prompt template
GENERATED = "generated" # LLM-generated artifact
SCHEMA = "schema" # Validation schema
CONFIG = "config" # Configuration artifact
@dataclass
class ArtifactMetadata:
"""
Extensible metadata for artifacts.
Attributes:
description: Human-readable description
tags: List of tags for categorization
author: Author identifier
version: Version string (optional)
custom: Dictionary for custom metadata fields
"""
description: Optional[str] = None
tags: list[str] = field(default_factory=list)
author: Optional[str] = None
version: Optional[str] = None
custom: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
"""Convert metadata to dictionary for serialization."""
return {
"description": self.description,
"tags": self.tags,
"author": self.author,
"version": self.version,
"custom": self.custom,
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "ArtifactMetadata":
"""Create metadata from dictionary."""
return cls(
description=data.get("description"),
tags=data.get("tags", []),
author=data.get("author"),
version=data.get("version"),
custom=data.get("custom", {}),
)
@dataclass
class Artifact:
"""
Core artifact entity with content-based addressing.
Implements FR-1: InformationSpace Addressability
- Unique persistent identifier
- Content digest for change detection
- Cross-space references via space_id
Attributes:
id: Unique identifier (UUID)
space_id: ID of containing InformationSpace
name: Human-readable name (unique within space)
artifact_type: Classification of artifact
content_digest: SHA-256 hash of content
content_size: Size of content in bytes
metadata: Extensible metadata
created_at: Creation timestamp
updated_at: Last modification timestamp
"""
id: str
space_id: str
name: str
artifact_type: ArtifactType
content_digest: str
content_size: int = 0
metadata: ArtifactMetadata = field(default_factory=ArtifactMetadata)
created_at: datetime = field(default_factory=datetime.utcnow)
updated_at: datetime = field(default_factory=datetime.utcnow)
@classmethod
def create(
cls,
space_id: str,
name: str,
content: str,
artifact_type: ArtifactType = ArtifactType.CONTENT,
metadata: Optional[ArtifactMetadata] = None,
) -> "Artifact":
"""
Create a new artifact with automatic ID and digest generation.
Args:
space_id: ID of containing space
name: Artifact name
content: Artifact content
artifact_type: Type classification
metadata: Optional metadata
Returns:
New Artifact instance with computed digest
"""
artifact_id = str(uuid.uuid4())
content_digest = calculate_content_digest(content)
content_size = len(content.encode('utf-8'))
return cls(
id=artifact_id,
space_id=space_id,
name=name,
artifact_type=artifact_type,
content_digest=content_digest,
content_size=content_size,
metadata=metadata or ArtifactMetadata(),
)
def update_content(self, new_content: str) -> None:
"""
Update artifact content and recalculate digest.
Args:
new_content: New content string
"""
self.content_digest = calculate_content_digest(new_content)
self.content_size = len(new_content.encode('utf-8'))
self.updated_at = datetime.utcnow()
def has_changed(self, current_digest: str) -> bool:
"""
Check if content has changed by comparing digests.
Args:
current_digest: Digest to compare against
Returns:
True if digests differ
"""
return self.content_digest != current_digest
def to_dict(self) -> Dict[str, Any]:
"""Convert artifact to dictionary for serialization."""
return {
"id": self.id,
"space_id": self.space_id,
"name": self.name,
"artifact_type": self.artifact_type.value,
"content_digest": self.content_digest,
"content_size": self.content_size,
"metadata": self.metadata.to_dict(),
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat(),
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "Artifact":
"""Create artifact from dictionary."""
return cls(
id=data["id"],
space_id=data["space_id"],
name=data["name"],
artifact_type=ArtifactType(data["artifact_type"]),
content_digest=data["content_digest"],
content_size=data.get("content_size", 0),
metadata=ArtifactMetadata.from_dict(data.get("metadata", {})),
created_at=datetime.fromisoformat(data["created_at"]),
updated_at=datetime.fromisoformat(data["updated_at"]),
)
@dataclass
class ArtifactReference:
"""
Reference to an artifact, possibly in another space.
Implements FR-1.3: Cross-space artifact references
Attributes:
name: Artifact name to resolve
space_id: Optional specific space ID (if not provided, uses resolution order)
version: Optional version constraint
"""
name: str
space_id: Optional[str] = None
version: Optional[str] = None
def __str__(self) -> str:
"""String representation for display and debugging."""
if self.space_id:
ref = f"{self.space_id}:{self.name}"
else:
ref = self.name
if self.version:
ref = f"{ref}@{self.version}"
return ref
@classmethod
def parse(cls, reference_str: str) -> "ArtifactReference":
"""
Parse a reference string into components.
Formats:
- "artifact-name"
- "space-id:artifact-name"
- "artifact-name@version"
- "space-id:artifact-name@version"
Args:
reference_str: Reference string to parse
Returns:
Parsed ArtifactReference
Raises:
ValueError: If reference string is malformed
"""
# Split version if present
if '@' in reference_str:
ref_part, version = reference_str.rsplit('@', 1)
else:
ref_part, version = reference_str, None
# Split space if present
if ':' in ref_part:
space_id, name = ref_part.split(':', 1)
else:
space_id, name = None, ref_part
if not name:
raise ValueError(f"Invalid artifact reference: {reference_str}")
return cls(name=name, space_id=space_id, version=version)
def calculate_content_digest(content: str) -> str:
"""
Calculate SHA-256 digest of content.
Implements FR-1.2: Content digest computation
Args:
content: Content string to hash
Returns:
Hexadecimal SHA-256 digest
"""
return hashlib.sha256(content.encode('utf-8')).hexdigest()
def calculate_bundle_digest(components: Dict[str, str]) -> str:
"""
Calculate digest of multiple components (for InputBundleHash).
Args:
components: Dictionary of component name -> digest
Returns:
Hexadecimal SHA-256 digest of sorted components
"""
# Sort components for deterministic hashing
sorted_items = sorted(components.items())
bundle_str = ':'.join(f"{k}={v}" for k, v in sorted_items)
return hashlib.sha256(bundle_str.encode('utf-8')).hexdigest()

20
markitect/prompts/repositories/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""
Repository layer for Prompt Dependency Resolution.
This package provides abstract interfaces and concrete implementations
for persisting prompts system entities.
"""
from markitect.prompts.repositories.interfaces import (
IArtifactRepository,
RepositoryError,
ArtifactNotFoundError,
DuplicateArtifactError,
)
__all__ = [
"IArtifactRepository",
"RepositoryError",
"ArtifactNotFoundError",
"DuplicateArtifactError",
]

159
markitect/prompts/repositories/interfaces.py Normal file
View File

@@ -0,0 +1,159 @@
"""
Repository interfaces for artifact persistence.
Defines abstract interfaces for artifact storage, enabling
pluggable storage backends.
"""
from abc import ABC, abstractmethod
from typing import List, Optional
from markitect.prompts.models import Artifact, ArtifactType
class RepositoryError(Exception):
"""Base exception for repository errors."""
pass
class ArtifactNotFoundError(RepositoryError):
"""Raised when an artifact cannot be found."""
pass
class DuplicateArtifactError(RepositoryError):
"""Raised when attempting to create an artifact with duplicate name."""
pass
class IArtifactRepository(ABC):
"""
Abstract interface for artifact persistence.
Implements FR-1: InformationSpace Addressability
Provides CRUD operations for artifacts with content digest tracking.
"""
@abstractmethod
def create(self, artifact: Artifact) -> Artifact:
"""
Persist a new artifact.
Args:
artifact: Artifact to create
Returns:
Created artifact
Raises:
DuplicateArtifactError: If artifact with same space_id+name exists
RepositoryError: On other persistence errors
"""
pass
@abstractmethod
def get_by_id(self, artifact_id: str) -> Optional[Artifact]:
"""
Retrieve artifact by ID.
Args:
artifact_id: Artifact identifier
Returns:
Artifact if found, None otherwise
"""
pass
@abstractmethod
def get_by_name(self, space_id: str, name: str) -> Optional[Artifact]:
"""
Retrieve artifact by space and name.
Implements FR-1.3: Cross-space artifact lookup
Args:
space_id: Space identifier
name: Artifact name
Returns:
Artifact if found, None otherwise
"""
pass
@abstractmethod
def get_by_digest(self, content_digest: str) -> List[Artifact]:
"""
Find artifacts with matching content digest.
Implements FR-1.2: Content digest queries
Args:
content_digest: SHA-256 digest to match
Returns:
List of artifacts with matching digest
"""
pass
@abstractmethod
def list_by_space(
self,
space_id: str,
artifact_type: Optional[ArtifactType] = None,
) -> List[Artifact]:
"""
List all artifacts in a space.
Args:
space_id: Space identifier
artifact_type: Optional type filter
Returns:
List of artifacts in space
"""
pass
@abstractmethod
def update(self, artifact: Artifact) -> Artifact:
"""
Update an existing artifact.
Updates content digest and modified timestamp.
Args:
artifact: Artifact with updated data
Returns:
Updated artifact
Raises:
ArtifactNotFoundError: If artifact doesn't exist
RepositoryError: On other persistence errors
"""
pass
@abstractmethod
def delete(self, artifact_id: str) -> bool:
"""
Delete an artifact.
Args:
artifact_id: Artifact identifier
Returns:
True if deleted, False if not found
"""
pass
@abstractmethod
def exists(self, space_id: str, name: str) -> bool:
"""
Check if artifact exists.
Args:
space_id: Space identifier
name: Artifact name
Returns:
True if artifact exists
"""
pass

332
markitect/prompts/repositories/sqlite.py Normal file
View File

@@ -0,0 +1,332 @@
"""
SQLite implementation of artifact repositories.
This module provides SQLite-backed implementations of the repository
interfaces for persistent storage of prompt artifacts.
"""
import sqlite3
import json
from pathlib import Path
from typing import List, Optional
from datetime import datetime
from markitect.prompts.repositories.interfaces import (
IArtifactRepository,
ArtifactNotFoundError,
DuplicateArtifactError,
RepositoryError,
)
from markitect.prompts.models import Artifact, ArtifactType, ArtifactMetadata
# SQL Schema for artifact tables
ARTIFACT_TABLES_SQL = """
-- Prompt artifacts table
CREATE TABLE IF NOT EXISTS prompt_artifacts (
id TEXT PRIMARY KEY,
space_id TEXT NOT NULL,
name TEXT NOT NULL,
artifact_type TEXT NOT NULL,
content_digest TEXT NOT NULL,
content_size INTEGER DEFAULT 0,
metadata JSON,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(space_id, name)
);
-- Indexes for performance
CREATE INDEX IF NOT EXISTS idx_artifacts_space ON prompt_artifacts(space_id);
CREATE INDEX IF NOT EXISTS idx_artifacts_digest ON prompt_artifacts(content_digest);
CREATE INDEX IF NOT EXISTS idx_artifacts_type ON prompt_artifacts(artifact_type);
CREATE INDEX IF NOT EXISTS idx_artifacts_name ON prompt_artifacts(space_id, name);
"""
def initialize_artifact_tables(db_path: str) -> None:
"""
Initialize the artifact-related database tables.
Args:
db_path: Path to the SQLite database file
"""
# Ensure directory exists
db_dir = Path(db_path).parent
if db_dir and not db_dir.exists():
db_dir.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(db_path)
try:
conn.executescript(ARTIFACT_TABLES_SQL)
conn.commit()
finally:
conn.close()
class SQLiteArtifactRepository(IArtifactRepository):
"""
SQLite implementation of artifact repository.
Provides persistent storage for artifacts with content digest tracking.
"""
def __init__(self, db_path: str):
"""
Initialize repository with database path.
Args:
db_path: Path to SQLite database file
"""
self.db_path = db_path
initialize_artifact_tables(db_path)
def _get_connection(self) -> sqlite3.Connection:
"""Get a database connection."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def create(self, artifact: Artifact) -> Artifact:
"""
Persist a new artifact.
Args:
artifact: Artifact to create
Returns:
Created artifact
Raises:
DuplicateArtifactError: If artifact with same space_id+name exists
RepositoryError: On other persistence errors
"""
conn = self._get_connection()
try:
conn.execute(
"""
INSERT INTO prompt_artifacts (
id, space_id, name, artifact_type, content_digest,
content_size, metadata, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
artifact.id,
artifact.space_id,
artifact.name,
artifact.artifact_type.value,
artifact.content_digest,
artifact.content_size,
json.dumps(artifact.metadata.to_dict()),
artifact.created_at.isoformat(),
artifact.updated_at.isoformat(),
),
)
conn.commit()
return artifact
except sqlite3.IntegrityError as e:
if "UNIQUE constraint" in str(e):
raise DuplicateArtifactError(
f"Artifact '{artifact.name}' already exists in space '{artifact.space_id}'"
)
raise RepositoryError(f"Database integrity error: {e}")
except Exception as e:
raise RepositoryError(f"Failed to create artifact: {e}")
finally:
conn.close()
def get_by_id(self, artifact_id: str) -> Optional[Artifact]:
"""
Retrieve artifact by ID.
Args:
artifact_id: Artifact identifier
Returns:
Artifact if found, None otherwise
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_artifacts WHERE id = ?",
(artifact_id,)
)
row = cursor.fetchone()
return self._row_to_artifact(row) if row else None
finally:
conn.close()
def get_by_name(self, space_id: str, name: str) -> Optional[Artifact]:
"""
Retrieve artifact by space and name.
Args:
space_id: Space identifier
name: Artifact name
Returns:
Artifact if found, None otherwise
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_artifacts WHERE space_id = ? AND name = ?",
(space_id, name)
)
row = cursor.fetchone()
return self._row_to_artifact(row) if row else None
finally:
conn.close()
def get_by_digest(self, content_digest: str) -> List[Artifact]:
"""
Find artifacts with matching content digest.
Args:
content_digest: SHA-256 digest to match
Returns:
List of artifacts with matching digest
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT * FROM prompt_artifacts WHERE content_digest = ?",
(content_digest,)
)
return [self._row_to_artifact(row) for row in cursor.fetchall()]
finally:
conn.close()
def list_by_space(
self,
space_id: str,
artifact_type: Optional[ArtifactType] = None,
) -> List[Artifact]:
"""
List all artifacts in a space.
Args:
space_id: Space identifier
artifact_type: Optional type filter
Returns:
List of artifacts in space
"""
conn = self._get_connection()
try:
if artifact_type:
cursor = conn.execute(
"SELECT * FROM prompt_artifacts WHERE space_id = ? AND artifact_type = ? ORDER BY name",
(space_id, artifact_type.value)
)
else:
cursor = conn.execute(
"SELECT * FROM prompt_artifacts WHERE space_id = ? ORDER BY name",
(space_id,)
)
return [self._row_to_artifact(row) for row in cursor.fetchall()]
finally:
conn.close()
def update(self, artifact: Artifact) -> Artifact:
"""
Update an existing artifact.
Args:
artifact: Artifact with updated data
Returns:
Updated artifact
Raises:
ArtifactNotFoundError: If artifact doesn't exist
RepositoryError: On other persistence errors
"""
conn = self._get_connection()
try:
# Update timestamp
artifact.updated_at = datetime.utcnow()
cursor = conn.execute(
"""
UPDATE prompt_artifacts
SET content_digest = ?, content_size = ?, metadata = ?, updated_at = ?
WHERE id = ?
""",
(
artifact.content_digest,
artifact.content_size,
json.dumps(artifact.metadata.to_dict()),
artifact.updated_at.isoformat(),
artifact.id,
),
)
if cursor.rowcount == 0:
raise ArtifactNotFoundError(f"Artifact with ID '{artifact.id}' not found")
conn.commit()
return artifact
except ArtifactNotFoundError:
raise
except Exception as e:
raise RepositoryError(f"Failed to update artifact: {e}")
finally:
conn.close()
def delete(self, artifact_id: str) -> bool:
"""
Delete an artifact.
Args:
artifact_id: Artifact identifier
Returns:
True if deleted, False if not found
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"DELETE FROM prompt_artifacts WHERE id = ?",
(artifact_id,)
)
conn.commit()
return cursor.rowcount > 0
finally:
conn.close()
def exists(self, space_id: str, name: str) -> bool:
"""
Check if artifact exists.
Args:
space_id: Space identifier
name: Artifact name
Returns:
True if artifact exists
"""
conn = self._get_connection()
try:
cursor = conn.execute(
"SELECT COUNT(*) FROM prompt_artifacts WHERE space_id = ? AND name = ?",
(space_id, name)
)
count = cursor.fetchone()[0]
return count > 0
finally:
conn.close()
def _row_to_artifact(self, row: sqlite3.Row) -> Artifact:
"""Convert database row to Artifact instance."""
metadata_dict = json.loads(row["metadata"]) if row["metadata"] else {}
return Artifact(
id=row["id"],
space_id=row["space_id"],
name=row["name"],
artifact_type=ArtifactType(row["artifact_type"]),
content_digest=row["content_digest"],
content_size=row["content_size"],
metadata=ArtifactMetadata.from_dict(metadata_dict),
created_at=datetime.fromisoformat(row["created_at"]),
updated_at=datetime.fromisoformat(row["updated_at"]),
)

10
markitect/prompts/services/__init__.py Normal file
View File

@@ -0,0 +1,10 @@
"""
Service layer for Prompt Dependency Resolution.
This package provides high-level business logic for artifact and template
management, orchestrating repositories and domain operations.
"""
from markitect.prompts.services.artifact_service import ArtifactService
__all__ = ["ArtifactService"]

239
markitect/prompts/services/artifact_service.py Normal file
View File

@@ -0,0 +1,239 @@
"""
Artifact service for high-level artifact management operations.
This service provides business logic for creating, querying, and updating
artifacts with automatic content digest calculation and change tracking.
"""
from typing import List, Optional
from datetime import datetime
from markitect.prompts.models import (
Artifact,
ArtifactType,
ArtifactMetadata,
ArtifactReference,
calculate_content_digest,
)
from markitect.prompts.repositories.interfaces import (
IArtifactRepository,
ArtifactNotFoundError,
)
class ArtifactService:
"""
Service for artifact management operations.
Provides high-level business logic on top of artifact repository,
handling content digest calculation, change detection, and cross-space
artifact resolution.
"""
def __init__(self, repository: IArtifactRepository):
"""
Initialize service with repository.
Args:
repository: Artifact repository implementation
"""
self.repository = repository
def create_artifact(
self,
space_id: str,
name: str,
content: str,
artifact_type: ArtifactType = ArtifactType.CONTENT,
metadata: Optional[ArtifactMetadata] = None,
) -> Artifact:
"""
Create a new artifact with automatic digest calculation.
Args:
space_id: ID of containing space
name: Artifact name
content: Artifact content
artifact_type: Type classification
metadata: Optional metadata
Returns:
Created artifact
Raises:
DuplicateArtifactError: If artifact already exists
"""
artifact = Artifact.create(
space_id=space_id,
name=name,
content=content,
artifact_type=artifact_type,
metadata=metadata,
)
return self.repository.create(artifact)
def get_artifact(self, artifact_id: str) -> Artifact:
"""
Retrieve artifact by ID.
Args:
artifact_id: Artifact identifier
Returns:
Artifact instance
Raises:
ArtifactNotFoundError: If artifact doesn't exist
"""
artifact = self.repository.get_by_id(artifact_id)
if not artifact:
raise ArtifactNotFoundError(f"Artifact with ID '{artifact_id}' not found")
return artifact
def get_artifact_by_name(self, space_id: str, name: str) -> Artifact:
"""
Retrieve artifact by space and name.
Args:
space_id: Space identifier
name: Artifact name
Returns:
Artifact instance
Raises:
ArtifactNotFoundError: If artifact doesn't exist
"""
artifact = self.repository.get_by_name(space_id, name)
if not artifact:
raise ArtifactNotFoundError(
f"Artifact '{name}' not found in space '{space_id}'"
)
return artifact
def resolve_reference(
self,
reference: ArtifactReference,
search_spaces: List[str],
) -> Optional[Artifact]:
"""
Resolve an artifact reference across multiple spaces.
Implements FR-1.3: Cross-space artifact references
Searches spaces in order until artifact is found.
Args:
reference: Artifact reference to resolve
search_spaces: List of space IDs to search in order
Returns:
Resolved artifact, or None if not found
"""
# If reference specifies space, search only that space
if reference.space_id:
return self.repository.get_by_name(reference.space_id, reference.name)
# Search spaces in order
for space_id in search_spaces:
artifact = self.repository.get_by_name(space_id, reference.name)
if artifact:
# TODO: Handle version constraint when versioning is implemented
return artifact
return None
def update_artifact_content(
self,
artifact_id: str,
new_content: str,
) -> Artifact:
"""
Update artifact content with automatic digest recalculation.
Args:
artifact_id: Artifact to update
new_content: New content
Returns:
Updated artifact
Raises:
ArtifactNotFoundError: If artifact doesn't exist
"""
artifact = self.get_artifact(artifact_id)
artifact.update_content(new_content)
return self.repository.update(artifact)
def detect_change(self, artifact_id: str, current_content: str) -> bool:
"""
Detect if artifact content has changed.
Args:
artifact_id: Artifact to check
current_content: Current content to compare
Returns:
True if content has changed
Raises:
ArtifactNotFoundError: If artifact doesn't exist
"""
artifact = self.get_artifact(artifact_id)
current_digest = calculate_content_digest(current_content)
return artifact.has_changed(current_digest)
def list_artifacts(
self,
space_id: str,
artifact_type: Optional[ArtifactType] = None,
) -> List[Artifact]:
"""
List artifacts in a space.
Args:
space_id: Space identifier
artifact_type: Optional type filter
Returns:
List of artifacts
"""
return self.repository.list_by_space(space_id, artifact_type)
def find_by_digest(self, content_digest: str) -> List[Artifact]:
"""
Find all artifacts with matching content digest.
Useful for finding duplicate content across spaces.
Args:
content_digest: SHA-256 digest to match
Returns:
List of artifacts with matching digest
"""
return self.repository.get_by_digest(content_digest)
def delete_artifact(self, artifact_id: str) -> bool:
"""
Delete an artifact.
Args:
artifact_id: Artifact to delete
Returns:
True if deleted, False if not found
"""
return self.repository.delete(artifact_id)
def artifact_exists(self, space_id: str, name: str) -> bool:
"""
Check if artifact exists.
Args:
space_id: Space identifier
name: Artifact name
Returns:
True if artifact exists
"""
return self.repository.exists(space_id, name)