"""Orchestration layer for prompt workflows.
:class:`PromptService` coordinates registration, alias movement, resolution,
rendering, and export across the persistence and storage layers. Both the
FastAPI API and the Rich CLI delegate to this service.
Most application code should use :class:`~promptdb.client.PromptClient`
instead — it wraps this service with ergonomic helpers for compact references,
file registration, and LangChain materialization.
Wiring a service manually (the client does this for you)::
from promptdb.db import create_all, create_session_factory
from promptdb.storage import LocalBlobStore
from promptdb.service import PromptService
create_all("sqlite:///./promptdb.sqlite3")
service = PromptService(
session_factory=create_session_factory("sqlite:///./promptdb.sqlite3"),
blob_store=LocalBlobStore(".blobs"),
)
Using the service::
from promptdb.domain import PromptRegistration, PromptSpec, PromptKind
version = service.register(PromptRegistration(
namespace="support", name="triage",
spec=PromptSpec(kind=PromptKind.STRING, template="Hi {name}"),
alias="production",
))
resolved = service.resolve(PromptRef.parse("support/triage:production"))
result = service.render(resolved.ref, {"name": "Will"})
"""
from __future__ import annotations
from pathlib import Path
from sqlalchemy.orm import Session, sessionmaker
from promptdb.db import PromptRepository
from promptdb.domain import (
PromptAssetKind,
PromptAssetView,
PromptRef,
PromptRegistration,
PromptRenderResult,
PromptVersionView,
)
from promptdb.storage import LocalBlobStore, MinioBlobStore, object_metadata
BlobStore = LocalBlobStore | MinioBlobStore
[docs]
class PromptService:
"""Application service for prompt workflows.
Args:
session_factory: SQLAlchemy session factory.
blob_store: Storage adapter with ``put_text`` and ``get_text``.
Returns:
PromptService: Service object.
Raises:
None.
Examples:
.. code-block:: python
service = PromptService(session_factory, blob_store)
"""
def __init__(self, session_factory: sessionmaker[Session], blob_store: BlobStore) -> None:
self.session_factory = session_factory
self.blob_store = blob_store
[docs]
def register(self, registration: PromptRegistration) -> PromptVersionView:
"""Register a new immutable prompt version.
Args:
registration: Registration payload.
Returns:
PromptVersionView: Created version.
Raises:
SQLAlchemyError: If persistence fails.
Examples:
.. code-block:: python
version = service.register(registration)
"""
with self.session_factory() as session:
repository = PromptRepository(session)
version = repository.create_version(
namespace=registration.namespace,
name=registration.name,
spec=registration.spec,
created_by=registration.created_by,
)
if registration.alias:
repository.move_alias(
namespace=registration.namespace,
name=registration.name,
alias=registration.alias,
version_id=version.id,
)
session.commit()
return repository.resolve(
namespace=registration.namespace,
name=registration.name,
selector=version.id,
)
[docs]
def move_alias(
self,
*,
namespace: str,
name: str,
alias: str,
version_id: str,
) -> PromptVersionView:
"""Move an alias and return the target version.
Args:
namespace: Prompt namespace.
name: Prompt name.
alias: Alias name.
version_id: Target version id.
Returns:
PromptVersionView: Target version.
Raises:
LookupError: If the prompt is missing.
Examples:
.. code-block:: python
view = service.move_alias(
namespace='support', name='triage',
alias='production', version_id='...',
)
"""
with self.session_factory() as session:
repository = PromptRepository(session)
repository.move_alias(
namespace=namespace, name=name, alias=alias, version_id=version_id
)
session.commit()
return repository.resolve(namespace=namespace, name=name, selector=version_id)
[docs]
def resolve(self, ref: PromptRef) -> PromptVersionView:
"""Resolve a prompt reference.
Args:
ref: Prompt reference.
Returns:
PromptVersionView: Resolved version.
Raises:
LookupError: If resolution fails.
Examples:
.. code-block:: python
view = service.resolve(PromptRef(namespace='support', name='triage'))
"""
with self.session_factory() as session:
repository = PromptRepository(session)
return repository.resolve(namespace=ref.namespace, name=ref.name, selector=ref.selector)
[docs]
def render(self, ref: PromptRef, variables: dict[str, object]) -> PromptRenderResult:
"""Resolve and render a prompt.
Args:
ref: Prompt reference.
variables: Runtime variables.
Returns:
PromptRenderResult: Rendered output.
Raises:
LookupError: If resolution fails.
Examples:
.. code-block:: python
ref = PromptRef(namespace='support', name='triage')
result = service.render(ref, {'question': 'hello'})
"""
version = self.resolve(ref)
return version.render(variables)
[docs]
def list_versions(self) -> list[PromptVersionView]:
"""List all known versions.
Args:
None.
Returns:
list[PromptVersionView]: Version views.
Raises:
None.
Examples:
.. code-block:: python
versions = service.list_versions()
"""
with self.session_factory() as session:
return PromptRepository(session).list_versions()
[docs]
def export_bundle(
self, version: PromptVersionView, *, key_prefix: str = "exports"
) -> PromptAssetView:
"""Export a prompt version bundle to blob storage.
Args:
version: Prompt version to export.
key_prefix: Storage key prefix.
Returns:
PromptAssetView: Relational asset view linked to the stored blob.
Raises:
OSError: If writing fails.
Examples:
.. code-block:: python
key = service.export_bundle(version)
"""
key = f"{key_prefix}/{version.namespace}/{version.name}/{version.version_id}.json"
payload = version.model_dump_json(indent=2)
self.blob_store.put_text(key, payload)
meta = object_metadata(
self.blob_store, key, content=payload, content_type="application/json"
)
with self.session_factory() as session:
repository = PromptRepository(session)
byte_size = meta["byte_size"]
checksum = meta["checksum_sha256"]
repository.create_asset(
version_id=version.version_id,
kind=PromptAssetKind.EXPORT_BUNDLE,
storage_backend=str(meta["storage_backend"]),
bucket=str(meta["bucket"]),
object_key=str(meta["object_key"]),
content_type="application/json",
byte_size=int(byte_size) if isinstance(byte_size, int) else None,
checksum_sha256=str(checksum) if checksum is not None else None,
metadata_json={
"namespace": version.namespace,
"name": version.name,
"revision": str(version.revision),
},
)
session.commit()
assets = repository.list_assets(version_id=version.version_id)
return assets[0]
[docs]
def list_assets(self, ref: PromptRef) -> list[PromptAssetView]:
"""List relational blob assets for a resolved prompt version.
Args:
ref: Prompt reference.
Returns:
list[PromptAssetView]: Linked asset metadata.
Raises:
LookupError: If the prompt cannot be resolved.
"""
version = self.resolve(ref)
with self.session_factory() as session:
return PromptRepository(session).list_assets(version_id=version.version_id)
[docs]
def export_to_file(self, version: PromptVersionView, path: str | Path) -> Path:
"""Export a prompt version to a local JSON file.
Args:
version: Prompt version.
path: Output path.
Returns:
Path: Output path.
Raises:
OSError: If writing fails.
Examples:
.. code-block:: python
service.export_to_file(version, 'build/version.json')
"""
output_path = Path(path)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(version.model_dump_json(indent=2), encoding="utf-8")
return output_path