db_prompts.py

• Pfad: /var/www/scripts/pipeline/db_prompts.py
• Namespace: pipeline
• Zeilen: 70 | Größe: 2,193 Bytes
• Geändert: 2025-12-28 08:57:12 | Gescannt: 2025-12-31 10:22:15

Code Hygiene Score: 100

• Dependencies: 100 (25%)
• LOC: 100 (20%)
• Methods: 100 (20%)
• Secrets: 100 (15%)
• Classes: 100 (10%)
• Magic Numbers: 100 (10%)

Keine Issues gefunden.

Klassen 1

• PromptsMixin class Zeile 8

Code

"""
Database Prompts Mixin

Single Responsibility: Prompt retrieval and management.
"""


class PromptsMixin:
    """Mixin for prompt operations.

    Provides:
    - get_prompt(): Get prompt by name
    - get_prompt_by_use_case(): Get prompt by use case with full metadata
    """

    def get_prompt(self, name: str, version: str = None) -> str | None:
        """Get a prompt by name (optionally specific version).

        Args:
            name: Prompt name
            version: Optional specific version

        Returns:
            Prompt content or None if not found
        """
        if version:
            cursor = self.execute(
                "SELECT content FROM prompts WHERE name = %s AND version = %s",
                (name, version),
            )
        else:
            cursor = self.execute(
                """SELECT content FROM prompts
                   WHERE name = %s AND is_active = 1
                   ORDER BY version DESC LIMIT 1""",
                (name,),
            )
        result = cursor.fetchone()
        cursor.close()
        return result["content"] if result else None

    def get_prompt_by_use_case(self, use_case: str, version: str = None) -> dict | None:
        """Get prompt by use_case with full metadata for provenance tracking.

        Args:
            use_case: The use case (entity_extraction, semantic_analysis, etc.)
            version: Optional specific version (otherwise latest active)

        Returns:
            Dict with id, name, version, content, use_case or None
        """
        if version:
            cursor = self.execute(
                """SELECT id, name, version, content, use_case
                   FROM prompts
                   WHERE use_case = %s AND version = %s""",
                (use_case, version),
            )
        else:
            cursor = self.execute(
                """SELECT id, name, version, content, use_case
                   FROM prompts
                   WHERE use_case = %s AND is_active = 1
                   ORDER BY version DESC LIMIT 1""",
                (use_case,),
            )
        result = cursor.fetchone()
        cursor.close()
        return result