API reference¶
Generated from the apogee-ai-prompt source with mkdocstrings. Every symbol below is exported from apogee_ai_prompt, so it is part of the supported public surface.
Application · DTOs¶
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
variables: list[PromptVariableDTO] = Field(default_factory=list)
Bases: BaseModel
Bases: BaseModel
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
versions: list[PromptVersionDTO] = Field(default_factory=list)
Bases: BaseModel
Bases: BaseModel
class-attribute
instance-attribute
¶
Python
variables: list[PromptVariableDTO] = Field(default_factory=list)
class-attribute
instance-attribute
¶
Bases: BaseModel
Bases: BaseModel
Adds a new version to an existing prompt.
class-attribute
instance-attribute
¶
Python
variables: list[PromptVariableDTO] = Field(default_factory=list)
Application · Use cases¶
Python
CreatePromptUseCase(repository: IPromptRepository)
Source code in apogee_ai_prompt/application/use_cases/create_prompt_use_case.py
async
¶
Python
execute(dto: CreatePromptDTO) -> PromptOutputDTO
Source code in apogee_ai_prompt/application/use_cases/create_prompt_use_case.py
Python
async def execute(self, dto: CreatePromptDTO) -> PromptOutputDTO:
if await self._repository.exists(dto.slug):
raise PromptAlreadyExistsException(dto.slug)
first_version = PromptVersion(
version=1,
body=dto.body,
variables=tuple(variable_from_dto(v) for v in dto.variables),
engine=dto.engine,
model=dto.model,
provider=dto.provider,
created_by=dto.created_by,
notes=dto.notes,
)
prompt = Prompt(
slug=dto.slug,
kind=dto.kind,
name=dto.name,
description=dto.description,
tags=tuple(dto.tags),
versions=(first_version,),
current_version=None,
)
saved = await self._repository.save(prompt)
return prompt_to_output_dto(saved)
Python
DeletePromptUseCase(repository: IPromptRepository)
Source code in apogee_ai_prompt/application/use_cases/delete_prompt_use_case.py
async
¶
Python
DiffPromptsUseCase(repository: IPromptRepository)
Source code in apogee_ai_prompt/application/use_cases/diff_prompts_use_case.py
async
¶
Python
execute(dto: DiffPromptsDTO) -> PromptDiff
Source code in apogee_ai_prompt/application/use_cases/diff_prompts_use_case.py
Python
async def execute(self, dto: DiffPromptsDTO) -> PromptDiff:
prompt = await self._repository.find(dto.slug)
if prompt is None:
raise PromptNotFoundException(dto.slug)
if not prompt.has_version(dto.from_version):
raise PromptVersionNotFoundException(dto.slug, dto.from_version)
if not prompt.has_version(dto.to_version):
raise PromptVersionNotFoundException(dto.slug, dto.to_version)
a = prompt.get_version(dto.from_version)
b = prompt.get_version(dto.to_version)
unified = "".join(
difflib.unified_diff(
a.body.splitlines(keepends=True),
b.body.splitlines(keepends=True),
fromfile=f"{dto.slug}@v{dto.from_version}",
tofile=f"{dto.slug}@v{dto.to_version}",
lineterm="",
)
)
a_vars = {v.name for v in a.variables}
b_vars = {v.name for v in b.variables}
return PromptDiff(
slug=dto.slug,
from_version=dto.from_version,
to_version=dto.to_version,
unified_diff=unified,
added_variables=tuple(sorted(b_vars - a_vars)),
removed_variables=tuple(sorted(a_vars - b_vars)),
body_changed=a.body != b.body,
)
Python
GetPromptUseCase(repository: IPromptRepository)
Source code in apogee_ai_prompt/application/use_cases/get_prompt_use_case.py
async
¶
Python
execute(slug: str) -> PromptOutputDTO
Python
LintPromptUseCase(repository: IPromptRepository, linter: IPromptLinter)
Source code in apogee_ai_prompt/application/use_cases/lint_prompt_use_case.py
async
¶
Python
execute(dto: LintPromptDTO) -> LintReport
Source code in apogee_ai_prompt/application/use_cases/lint_prompt_use_case.py
Python
async def execute(self, dto: LintPromptDTO) -> LintReport:
prompt = await self._repository.find(dto.slug)
if prompt is None:
raise PromptNotFoundException(dto.slug)
version_number = dto.version
if version_number is None:
latest = prompt.latest_version()
if latest is None:
raise PromptVersionNotFoundException(dto.slug, 0)
version_number = latest.version
if not prompt.has_version(version_number):
raise PromptVersionNotFoundException(dto.slug, version_number)
version = prompt.get_version(version_number)
return self._linter.lint(dto.slug, version)
Python
ListPromptsUseCase(repository: IPromptRepository)
Source code in apogee_ai_prompt/application/use_cases/list_prompts_use_case.py
async
¶
Python
execute(filt: ListPromptsFilterDTO | None = None) -> list[PromptOutputDTO]
Source code in apogee_ai_prompt/application/use_cases/list_prompts_use_case.py
Python
PublishPromptUseCase(repository: IPromptRepository, linter: IPromptLinter | None = None)
Source code in apogee_ai_prompt/application/use_cases/publish_prompt_use_case.py
async
¶
Python
execute(dto: PublishPromptDTO) -> PromptOutputDTO
Source code in apogee_ai_prompt/application/use_cases/publish_prompt_use_case.py
Python
async def execute(self, dto: PublishPromptDTO) -> PromptOutputDTO:
prompt = await self._repository.find(dto.slug)
if prompt is None:
raise PromptNotFoundException(dto.slug)
if not prompt.has_version(dto.version):
raise PromptVersionNotFoundException(dto.slug, dto.version)
version = prompt.get_version(dto.version)
if self._linter is not None and not dto.skip_lint:
report = self._linter.lint(dto.slug, version)
if report.has_errors:
error_count = sum(1 for i in report.issues if i.severity.value == "error")
raise LintBlockedException(dto.slug, dto.version, error_count)
published = version.with_status(PromptStatus.PUBLISHED)
updated = prompt.with_replaced_version(published)
if dto.set_current:
updated = updated.with_current_version(dto.version)
saved = await self._repository.save(updated)
return prompt_to_output_dto(saved)
Python
RenderPromptUseCase(repository: IPromptRepository, engines: Mapping[TemplateEngineKind, ITemplateEngine], *, canary_configs: Mapping[str, CanaryConfig] | None = None, variant_resolver: IVariantResolver | None = None)
Render a Prompt by slug+version, optionally routed via canary.
Source code in apogee_ai_prompt/application/use_cases/render_prompt_use_case.py
Python
def __init__(
self,
repository: IPromptRepository,
engines: Mapping[TemplateEngineKind, ITemplateEngine],
*,
canary_configs: Mapping[str, CanaryConfig] | None = None,
variant_resolver: IVariantResolver | None = None,
) -> None:
self._repository = repository
self._engines = dict(engines)
self._canary = dict(canary_configs or {})
self._resolver = variant_resolver
async
¶
Python
execute(dto: RenderPromptDTO) -> RenderedPrompt
Source code in apogee_ai_prompt/application/use_cases/render_prompt_use_case.py
Python
async def execute(self, dto: RenderPromptDTO) -> RenderedPrompt:
prompt = await self._repository.find(dto.slug)
if prompt is None:
raise PromptNotFoundException(dto.slug)
version = self._select_version(prompt, dto)
coerced = validate_and_coerce_variables(version, dto.variables, strict=dto.strict)
engine = self._engines.get(version.engine)
if engine is None:
raise KeyError(f"No template engine registered for {version.engine.value!r}")
text = engine.render(version.body, coerced)
return RenderedPrompt(
slug=dto.slug,
version=version.version,
text=text,
variables_used=coerced,
engine=engine.name,
)
Python
ResolveVariantUseCase(resolver: IVariantResolver)
Source code in apogee_ai_prompt/application/use_cases/resolve_variant_use_case.py
Python
execute(config: CanaryConfig, *, routing_value: str) -> VariantSelection
Python
RollbackPromptUseCase(repository: IPromptRepository)
Sets current_version to a previously published version.
Source code in apogee_ai_prompt/application/use_cases/rollback_prompt_use_case.py
async
¶
Python
execute(dto: RollbackPromptDTO) -> PromptOutputDTO
Source code in apogee_ai_prompt/application/use_cases/rollback_prompt_use_case.py
Python
async def execute(self, dto: RollbackPromptDTO) -> PromptOutputDTO:
prompt = await self._repository.find(dto.slug)
if prompt is None:
raise PromptNotFoundException(dto.slug)
if not prompt.has_version(dto.to_version):
raise PromptVersionNotFoundException(dto.slug, dto.to_version)
updated = prompt.with_current_version(dto.to_version)
saved = await self._repository.save(updated)
return prompt_to_output_dto(saved)
Python
UpdatePromptUseCase(repository: IPromptRepository)
Adds a new DRAFT version to an existing prompt.
Source code in apogee_ai_prompt/application/use_cases/update_prompt_use_case.py
async
¶
Python
execute(dto: UpdatePromptDTO) -> PromptOutputDTO
Source code in apogee_ai_prompt/application/use_cases/update_prompt_use_case.py
Python
async def execute(self, dto: UpdatePromptDTO) -> PromptOutputDTO:
prompt = await self._repository.find(dto.slug)
if prompt is None:
raise PromptNotFoundException(dto.slug)
latest = prompt.latest_version()
engine = dto.engine or (latest.engine if latest else TemplateEngineKind.JINJA)
new_version = PromptVersion(
version=prompt.next_version_number(),
body=dto.body,
variables=tuple(variable_from_dto(v) for v in dto.variables),
engine=engine,
model=dto.model or (latest.model if latest else None),
provider=dto.provider or (latest.provider if latest else None),
created_by=dto.created_by,
notes=dto.notes,
)
updated = prompt.with_added_version(new_version)
saved = await self._repository.save(updated)
return prompt_to_output_dto(saved)
Domain¶
dataclass
¶
Weighted distribution of versions for a canary release.
variants maps version number → weight (must sum to 1.0). The resolver
uses a deterministic hash of the routing key (user_id / tenant_id / etc.)
to assign requests to versions, so the same key always lands in the same
bucket.
dataclass
¶
Python
CostEstimate(provider: str, model: str, input_tokens: int, output_tokens: int, input_cost_usd: float, output_cost_usd: float)
Cost estimate in USD given input/output token assumptions.
dataclass
¶
Python
LintIssue(rule: str, severity: LintSeverity, message: str, line: int | None = None)
dataclass
¶
Python
LintReport(slug: str, version: int, issues: tuple[LintIssue, ...] = tuple())
class-attribute
instance-attribute
¶
Python
issues: tuple[LintIssue, ...] = field(default_factory=tuple)
Python
of_severity(severity: LintSeverity) -> tuple[LintIssue, ...]
dataclass
¶
Python
Prompt(slug: str, kind: PromptKind = CHAT, name: str | None = None, description: str | None = None, tags: tuple[str, ...] = tuple(), versions: tuple[PromptVersion, ...] = tuple(), current_version: int | None = None, created_at: datetime = (lambda: now(utc))(), updated_at: datetime = (lambda: now(utc))())
Aggregate root: a named prompt identified by slug with N versions.
The prompt itself is a thin shell carrying the slug, kind, descriptive metadata and the ordered tuple of versions. Business rules around publishing, rolling back and version pinning are enforced at this level.
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
versions: tuple[PromptVersion, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
get_version(version: int) -> PromptVersion
Python
latest_version() -> PromptVersion | None
Python
published_versions() -> tuple[PromptVersion, ...]
Python
latest_published() -> PromptVersion | None
Python
with_added_version(new_version: PromptVersion) -> Prompt
Source code in apogee_ai_prompt/domain/entities/prompt.py
Python
def with_added_version(self, new_version: PromptVersion) -> Prompt:
if any(v.version == new_version.version for v in self.versions):
raise ValueError(
f"Version {new_version.version} already exists in prompt {self.slug!r}"
)
new_versions = tuple(sorted([*self.versions, new_version], key=lambda v: v.version))
return replace(
self,
versions=new_versions,
updated_at=datetime.now(timezone.utc),
)
Python
with_replaced_version(replacement: PromptVersion) -> Prompt
Source code in apogee_ai_prompt/domain/entities/prompt.py
Python
def with_replaced_version(self, replacement: PromptVersion) -> Prompt:
if not any(v.version == replacement.version for v in self.versions):
raise KeyError(f"Version {replacement.version} not found")
new_versions = tuple(
replacement if v.version == replacement.version else v for v in self.versions
)
return replace(
self,
versions=new_versions,
updated_at=datetime.now(timezone.utc),
)
dataclass
¶
Python
PromptDiff(slug: str, from_version: int, to_version: int, unified_diff: str, added_variables: tuple[str, ...] = (), removed_variables: tuple[str, ...] = (), body_changed: bool = True)
class-attribute
instance-attribute
¶
Bases: str, Enum
dataclass
¶
Python
PromptVariable(name: str, type: VariableType, required: bool = True, default: Any = None, description: str | None = None)
Schema declaration for a single variable consumed by a PromptVersion.
dataclass
¶
Python
PromptVersion(version: int, body: str, variables: tuple[PromptVariable, ...] = tuple(), engine: TemplateEngineKind = JINJA, model: str | None = None, provider: str | None = None, status: PromptStatus = DRAFT, created_at: datetime = (lambda: now(utc))(), created_by: str | None = None, notes: str | None = None, metadata: dict[str, str] = dict())
Single immutable version of a Prompt.
A PromptVersion captures the complete state of a prompt at a point in time: template body, schema of variables, target model/provider, render engine, plus status and authorship metadata. Once status reaches PUBLISHED it must not be mutated — new edits create a new version.
class-attribute
instance-attribute
¶
Python
variables: tuple[PromptVariable, ...] = field(default_factory=tuple)
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
Python
with_status(status: PromptStatus) -> PromptVersion
Source code in apogee_ai_prompt/domain/entities/prompt_version.py
Python
def with_status(self, status: PromptStatus) -> PromptVersion:
return PromptVersion(
version=self.version,
body=self.body,
variables=self.variables,
engine=self.engine,
model=self.model,
provider=self.provider,
status=status,
created_at=self.created_at,
created_by=self.created_by,
notes=self.notes,
metadata=self.metadata,
)
dataclass
¶
Python
RenderedPrompt(slug: str, version: int, text: str, variables_used: dict[str, Any] = dict(), engine: str = 'jinja')
class-attribute
instance-attribute
¶
dataclass
¶
dataclass
¶
Domain · Enums¶
Bases: str, Enum
Bases: str, Enum
Domain · Exceptions¶
Bases: PromptError
Source code in apogee_ai_prompt/domain/exceptions/prompt_exceptions.py
Bases: Exception
Base for all apogee-ai-prompt errors.
Bases: PromptError
Source code in apogee_ai_prompt/domain/exceptions/prompt_exceptions.py
Bases: PromptError
Source code in apogee_ai_prompt/domain/exceptions/prompt_exceptions.py
Domain · Protocols (ports)¶
Bases: Protocol
Python
estimate(*, provider: str, model: str, input_tokens: int, output_tokens: int) -> CostEstimate
Bases: Protocol
Python
lint(slug: str, version: PromptVersion) -> LintReport
Bases: Protocol
Combined Query + Command repository for Prompt aggregates.
Implementations may be read-only (GitPromptRepository reading from a
fixed ref) — those raise RepositoryReadOnlyException on writes.
Bases: Protocol
Python
count(text: str, *, model: str | None = None) -> TokenCount
Bases: Protocol
Python
resolve(config: CanaryConfig, *, routing_value: str) -> VariantSelection
Infrastructure¶
12-rule linter for prompt versions.
Rules:
empty-body(error): body must not be emptyinjection-pattern(warning): looks like a prompt-injection attempttoo-long(warning): body length abovemax_charstoo-short(info): body length belowmin_charsunused-variable(warning): declared but not referencedundeclared-variable(error): referenced in body but missing in schemarequired-with-default(info):required=Truebut a default is setmixed-engines(warning): body looks Jinja but engine is plain/mustachetrailing-whitespace(info): trailing whitespace linestriple-blank-line(info): three or more consecutive blank linesno-newline-at-end(info): missing terminal newlineno-system-prologue(info): chat prompt without leading role-setting line
Source code in apogee_ai_prompt/infrastructure/linters/default_prompt_linter.py
Python
lint(slug: str, version: PromptVersion) -> LintReport
Source code in apogee_ai_prompt/infrastructure/linters/default_prompt_linter.py
Python
def lint(self, slug: str, version: PromptVersion) -> LintReport:
issues: list[LintIssue] = []
body = version.body
# 1. empty-body
if not body or not body.strip():
issues.append(
LintIssue("empty-body", LintSeverity.ERROR, "Prompt body is empty")
)
# 2. injection-pattern
for pattern, rule_id in _INJECTION_PATTERNS:
for match in re.finditer(pattern, body):
issues.append(
LintIssue(
f"injection-pattern:{rule_id}",
LintSeverity.WARNING,
f"Possible prompt injection pattern: {match.group(0)!r}",
)
)
# 3 / 4. length
if len(body) > self._max_chars:
issues.append(
LintIssue(
"too-long",
LintSeverity.WARNING,
f"Body has {len(body)} chars (>{self._max_chars})",
)
)
if 0 < len(body.strip()) < self._min_chars:
issues.append(
LintIssue(
"too-short",
LintSeverity.INFO,
f"Body has only {len(body.strip())} chars (<{self._min_chars})",
)
)
# 5 / 6. variable consistency
try:
referenced = self._discover_for(version, body)
except Exception: # noqa: BLE001 - linter should never crash on bad input
referenced = set()
declared = {v.name for v in version.variables}
for name in sorted(declared - referenced):
issues.append(
LintIssue(
"unused-variable",
LintSeverity.WARNING,
f"Variable {name!r} declared but not used in body",
)
)
for name in sorted(referenced - declared):
issues.append(
LintIssue(
"undeclared-variable",
LintSeverity.ERROR,
f"Variable {name!r} used in body but not declared in schema",
)
)
# 7. required-with-default
for var in version.variables:
if var.required and var.default is not None:
issues.append(
LintIssue(
"required-with-default",
LintSeverity.INFO,
f"Variable {var.name!r} is required but also has a default",
)
)
# 8. mixed-engines
if version.engine != TemplateEngineKind.JINJA and re.search(r"\{\{.*?\}\}", body):
issues.append(
LintIssue(
"mixed-engines",
LintSeverity.WARNING,
f"Body uses Jinja-like {{{{ ... }}}} but engine is {version.engine.value}",
)
)
# 9. trailing whitespace
for i, line in enumerate(body.splitlines(), start=1):
if line and line != line.rstrip():
issues.append(
LintIssue(
"trailing-whitespace",
LintSeverity.INFO,
"Trailing whitespace",
line=i,
)
)
break
# 10. triple-blank-line
if "\n\n\n" in body:
issues.append(
LintIssue(
"triple-blank-line",
LintSeverity.INFO,
"Three or more consecutive blank lines",
)
)
# 11. no-newline-at-end
if body and not body.endswith("\n"):
issues.append(
LintIssue(
"no-newline-at-end",
LintSeverity.INFO,
"Body does not end with a newline",
)
)
# 12. no-system-prologue (only for chat prompts)
from ...domain.enums.prompt_kind_enum import PromptKind
kind_str = getattr(version, "kind", None)
del kind_str
# The version itself has no kind — chat-vs-other is at Prompt level.
# We still surface this as an INFO since it is a soft style suggestion
# only when the body is short and lacks a leading role definition.
first_line = body.strip().splitlines()[0].lower() if body.strip() else ""
if (
first_line
and len(body.strip()) < 200
and not re.match(
r"^(you are|act as|behave as|role:|system:)",
first_line,
)
):
issues.append(
LintIssue(
"no-system-prologue",
LintSeverity.INFO,
"Short prompt missing role-setting prologue ('You are ...').",
)
)
# Reference PromptKind to keep the import meaningful if reused.
_ = PromptKind
return LintReport(slug=slug, version=version.version, issues=tuple(issues))
YAML-on-disk repository: one <root>/<slug>.yml file per Prompt.
Source code in apogee_ai_prompt/infrastructure/repositories/file_prompt_repository.py
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/file_prompt_repository.py
Python
async def list(
self,
*,
tag: str | None = None,
limit: int | None = None,
offset: int = 0,
) -> list[Prompt]:
items = await asyncio.to_thread(self._read_all)
if tag is not None:
items = [p for p in items if tag in p.tags]
items.sort(key=lambda p: p.slug)
if offset:
items = items[offset:]
if limit is not None:
items = items[:limit]
return items
async
¶
async
¶
async
¶
Python
GitPromptRepository(repo_path: str | Path, *, ref: str = 'HEAD', prompts_dir: str = 'prompts')
Read-only repository that loads prompts from a fixed git ref.
Powered by dulwich — pure-Python git, so no system git binary is
required. Useful when prompt evolution must be auditable: production
pins a tag (v1.4.0) and any change requires a code review.
Writes raise RepositoryReadOnlyException.
Source code in apogee_ai_prompt/infrastructure/repositories/git_prompt_repository.py
Python
def __init__(
self,
repo_path: str | Path,
*,
ref: str = "HEAD",
prompts_dir: str = "prompts",
) -> None:
try:
from dulwich import porcelain # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"GitPromptRepository requires `dulwich`. "
"Install with: pip install 'apogee-ai-prompt[git]'"
) from exc
self._repo_path = Path(repo_path)
self._ref = ref
self._prompts_dir = prompts_dir.strip("/")
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/git_prompt_repository.py
Python
async def list(
self,
*,
tag: str | None = None,
limit: int | None = None,
offset: int = 0,
) -> list[Prompt]:
items = await asyncio.to_thread(self._read_all)
if tag is not None:
items = [p for p in items if tag in p.tags]
items.sort(key=lambda p: p.slug)
if offset:
items = items[offset:]
if limit is not None:
items = items[:limit]
return items
async
¶
async
¶
async
¶
Approximates token count as ceil(len(text)/4).
Decent for English-like text; not a substitute for tiktoken when accurate
counts matter (use TiktokenTokenCounter instead).
Python
count(text: str, *, model: str | None = None) -> TokenCount
Source code in apogee_ai_prompt/infrastructure/token_counters/heuristic_token_counter.py
Thread-unsafe in-process implementation; ideal for tests and dev runs.
Source code in apogee_ai_prompt/infrastructure/repositories/in_memory_prompt_repository.py
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/in_memory_prompt_repository.py
Python
async def list(
self,
*,
tag: str | None = None,
limit: int | None = None,
offset: int = 0,
) -> list[Prompt]:
items = list(self._store.values())
if tag is not None:
items = [p for p in items if tag in p.tags]
items.sort(key=lambda p: p.slug)
if offset:
items = items[offset:]
if limit is not None:
items = items[:limit]
return items
async
¶
async
¶
async
¶
Source code in apogee_ai_prompt/infrastructure/template_engines/jinja_template_engine.py
Source code in apogee_ai_prompt/infrastructure/template_engines/jinja_template_engine.py
Python
def render(self, body: str, variables: dict[str, Any]) -> str:
try:
template = self._env.from_string(body)
return template.render(**variables)
except UndefinedError as exc:
raise TemplateRenderException(f"Undefined variable in Jinja template: {exc}") from exc
except TemplateError as exc:
raise TemplateRenderException(f"Jinja render error: {exc}") from exc
module-attribute
¶
Python
PRICING_TABLE: dict[str, dict[str, tuple[float, float]]] = {'openai': {'gpt-4o': (5.0, 15.0), 'gpt-4o-mini': (0.15, 0.6), 'gpt-4-turbo': (10.0, 30.0), 'gpt-4': (30.0, 60.0), 'gpt-3.5-turbo': (0.5, 1.5), 'o1': (15.0, 60.0), 'o1-mini': (3.0, 12.0), 'o3-mini': (1.1, 4.4), 'text-embedding-3-large': (0.13, 0.0), 'text-embedding-3-small': (0.02, 0.0)}, 'anthropic': {'claude-3-5-sonnet': (3.0, 15.0), 'claude-3-5-haiku': (0.8, 4.0), 'claude-3-opus': (15.0, 75.0), 'claude-3-sonnet': (3.0, 15.0), 'claude-3-haiku': (0.25, 1.25), 'claude-opus-4': (15.0, 75.0), 'claude-sonnet-4': (3.0, 15.0), 'claude-haiku-4': (0.8, 4.0)}, 'google': {'gemini-1.5-pro': (1.25, 5.0), 'gemini-1.5-flash': (0.075, 0.3), 'gemini-2.0-flash': (0.1, 0.4), 'gemini-2.5-pro': (1.25, 5.0), 'gemini-2.5-flash': (0.1, 0.4)}, 'openrouter': {'default': (1.0, 3.0)}, 'bedrock': {'anthropic.claude-3-5-sonnet': (3.0, 15.0), 'anthropic.claude-3-haiku': (0.25, 1.25), 'amazon.titan-text-express': (0.2, 0.6), 'meta.llama3-70b-instruct': (2.65, 3.5)}}
Python str.format-style placeholder renderer with no extras.
Useful when neither Jinja nor Mustache is desired and you just want
{name} substitution.
Source code in apogee_ai_prompt/infrastructure/template_engines/plain_template_engine.py
Python
def render(self, body: str, variables: dict[str, Any]) -> str:
try:
return body.format(**variables)
except KeyError as exc:
raise TemplateRenderException(f"Missing variable in plain template: {exc}") from exc
except (IndexError, ValueError) as exc:
raise TemplateRenderException(f"Plain render error: {exc}") from exc
Source code in apogee_ai_prompt/infrastructure/template_engines/plain_template_engine.py
SQLAlchemy 2.x async repository.
The schema is intentionally simple: one row per Prompt with a JSON column storing the serialized aggregate. Lower friction for migrations than splitting versions into a separate table; trade-off is querying by version across prompts requires JSON path operators (Postgres) or scanning.
SQLAlchemy is imported lazily so installing apogee-ai-prompt without the
[sql] extra still works.
Source code in apogee_ai_prompt/infrastructure/repositories/sql_prompt_repository.py
Python
def __init__(self, session_factory: Any, *, table_name: str = "apogee_prompts") -> None:
try:
import sqlalchemy # type: ignore # noqa: F401
except ImportError as exc: # pragma: no cover - exercised by extras
raise ImportError(
"SqlPromptRepository requires SQLAlchemy. "
"Install with: pip install 'apogee-ai-prompt[sql]'"
) from exc
self._session_factory = session_factory
self._table_name = table_name
self._table = self._build_table()
async
¶
async
¶
Python
find(slug: str) -> Prompt | None
Source code in apogee_ai_prompt/infrastructure/repositories/sql_prompt_repository.py
Python
async def find(self, slug: str) -> Prompt | None:
from sqlalchemy import select
async with self._session_factory() as session:
stmt = select(self._table.c.data).where(self._table.c.slug == slug)
row = (await session.execute(stmt)).first()
if row is None:
return None
data = self._decode(row[0])
return prompt_from_dict(data)
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/sql_prompt_repository.py
Python
async def list(
self,
*,
tag: str | None = None,
limit: int | None = None,
offset: int = 0,
) -> list[Prompt]:
from sqlalchemy import select
async with self._session_factory() as session:
stmt = select(self._table.c.data).order_by(self._table.c.slug)
rows = (await session.execute(stmt)).all()
items = [prompt_from_dict(self._decode(r[0])) for r in rows]
if tag is not None:
items = [p for p in items if tag in p.tags]
if offset:
items = items[offset:]
if limit is not None:
items = items[:limit]
return items
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/sql_prompt_repository.py
Python
async def save(self, prompt: Prompt) -> Prompt:
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.exc import CompileError
payload = prompt_to_dict(prompt)
async with self._session_factory() as session:
try:
stmt = sqlite_insert(self._table).values(slug=prompt.slug, data=payload)
stmt = stmt.on_conflict_do_update(
index_elements=[self._table.c.slug],
set_={"data": payload},
)
await session.execute(stmt)
except (CompileError, NotImplementedError):
# Fallback: delete + insert (works on any backend)
from sqlalchemy import delete, insert
await session.execute(delete(self._table).where(self._table.c.slug == prompt.slug))
await session.execute(
insert(self._table).values(slug=prompt.slug, data=payload)
)
await session.commit()
return prompt
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/sql_prompt_repository.py
async
¶
Source code in apogee_ai_prompt/infrastructure/repositories/sql_prompt_repository.py
Cost estimator backed by a static pricing table.
Pricing is expressed per 1M tokens. Override or extend the bundled
PRICING_TABLE by passing a custom dict.
Source code in apogee_ai_prompt/infrastructure/cost_estimators/table_cost_estimator.py
Python
estimate(*, provider: str, model: str, input_tokens: int, output_tokens: int) -> CostEstimate
Source code in apogee_ai_prompt/infrastructure/cost_estimators/table_cost_estimator.py
Python
def estimate(
self,
*,
provider: str,
model: str,
input_tokens: int,
output_tokens: int,
) -> CostEstimate:
price = lookup_price(provider, model) if self._pricing is PRICING_TABLE else self._lookup(
provider, model
)
if price is None:
input_per_m, output_per_m = 0.0, 0.0
else:
input_per_m, output_per_m = price
input_cost = (input_tokens / 1_000_000.0) * input_per_m
output_cost = (output_tokens / 1_000_000.0) * output_per_m
return CostEstimate(
provider=provider,
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
input_cost_usd=round(input_cost, 6),
output_cost_usd=round(output_cost, 6),
)
Accurate counter backed by OpenAI's tiktoken.
Lazy import: only required if you install apogee-ai-prompt[tiktoken].
Source code in apogee_ai_prompt/infrastructure/token_counters/tiktoken_token_counter.py
Python
def __init__(self, *, default_encoding: str = "cl100k_base") -> None:
try:
import tiktoken # type: ignore # noqa: F401
except ImportError as exc:
raise ImportError(
"TiktokenTokenCounter requires `tiktoken`. "
"Install with: pip install 'apogee-ai-prompt[tiktoken]'"
) from exc
self._default_encoding = default_encoding
Python
count(text: str, *, model: str | None = None) -> TokenCount
Source code in apogee_ai_prompt/infrastructure/token_counters/tiktoken_token_counter.py
Python
def count(self, text: str, *, model: str | None = None) -> TokenCount:
import tiktoken # type: ignore
if model:
try:
enc = tiktoken.encoding_for_model(model)
except (KeyError, ValueError):
enc = tiktoken.get_encoding(self._default_encoding)
else:
enc = tiktoken.get_encoding(self._default_encoding)
return TokenCount(tokens=len(enc.encode(text)), counter=self.name)
Deterministic A/B + canary resolver.
Computes bucket = hash(slug:routing_value) / 2^64 (uniform in [0,1)) and
walks the variants in ascending version order, accumulating weights. The
first cumulative weight greater than the bucket wins. Same routing value
always lands on the same version, which is what makes this safe for
persistent A/B testing.
Python
resolve(config: CanaryConfig, *, routing_value: str) -> VariantSelection
Source code in apogee_ai_prompt/infrastructure/resolvers/weighted_hash_resolver.py
Python
def resolve(
self,
config: CanaryConfig,
*,
routing_value: str,
) -> VariantSelection:
digest = hashlib.sha256(f"{config.slug}:{routing_value}".encode()).digest()
# Use first 8 bytes as unsigned 64-bit integer
as_int = int.from_bytes(digest[:8], "big")
bucket = as_int / float(1 << 64)
cumulative = 0.0
for version in sorted(config.variants):
cumulative += config.variants[version]
if bucket < cumulative:
return VariantSelection(
slug=config.slug,
selected_version=version,
routing_key=config.routing_key,
routing_value=routing_value,
bucket=bucket,
)
# Fallback (rounding edge case): pick the last version
last = max(config.variants)
return VariantSelection(
slug=config.slug,
selected_version=last,
routing_key=config.routing_key,
routing_value=routing_value,
bucket=bucket,
)
Python
default_engines(*, include_mustache: bool = False) -> dict[TemplateEngineKind, ITemplateEngine]
Return a registry of template engines suitable for RenderPromptUseCase.
Mustache is opt-in because chevron is an optional dependency.
Source code in apogee_ai_prompt/infrastructure/template_engines/registry.py
Python
def default_engines(*, include_mustache: bool = False) -> dict[TemplateEngineKind, ITemplateEngine]:
"""Return a registry of template engines suitable for ``RenderPromptUseCase``.
Mustache is opt-in because ``chevron`` is an optional dependency.
"""
engines: dict[TemplateEngineKind, ITemplateEngine] = {
TemplateEngineKind.JINJA: JinjaTemplateEngine(),
TemplateEngineKind.PLAIN: PlainTemplateEngine(),
}
if include_mustache:
from .mustache_template_engine import MustacheTemplateEngine
engines[TemplateEngineKind.MUSTACHE] = MustacheTemplateEngine()
return engines
Return (price_input, price_output) per 1M tokens or None.
The lookup is case-insensitive. If the exact model is not registered the
provider's default entry (when present) is used.
Source code in apogee_ai_prompt/infrastructure/cost_estimators/pricing_table.py
Python
def lookup_price(provider: str, model: str) -> tuple[float, float] | None:
"""Return ``(price_input, price_output)`` per 1M tokens or ``None``.
The lookup is case-insensitive. If the exact model is not registered the
provider's ``default`` entry (when present) is used.
"""
p = provider.lower()
if p not in PRICING_TABLE:
return None
catalog = PRICING_TABLE[p]
m = model.lower()
if m in catalog:
return catalog[m]
# Substring fallback (e.g., "claude-3-5-sonnet-20241022" → "claude-3-5-sonnet")
for key in catalog:
if key in m:
return catalog[key]
if "default" in catalog:
return catalog["default"]
return None