teilchensammler-cli/src/teilchensammler_cli/models.py

import logging
import uuid

from sqlmodel import (
    Field,
    SQLModel,
    Session,
    select,
    Sequence,
)

logger = logging.getLogger(__name__)


class TeilchenCreate(SQLModel):
    description: str | None
    name: str = Field(index=True)
    number: int = Field(default=1)
    text: str
    tags: str | None


class Teilchen(TeilchenCreate, table=True):
    id: uuid.UUID = Field(default_factory=uuid.uuid7, primary_key=True)  # ty:ignore[unresolved-attribute]


async def make_teilchen_input(text: str) -> TeilchenCreate | None:
    """Constructor to create new Teilchen from user input.

    Args:
        text: The whole input string as provided by the user.

    Returns:
        Returns new Teilchen instance if enough parts could be extracted.
        Returns `None` otherwise.
    """
    import re
    from natsort import natsorted

    text = text.strip()

    if not text:
        logger.error("Empty text.")
        return None

    name_end = text.find(".")
    if name_end == -1:  # "." not in text
        logger.warning("Missing period '.' from text.")
        logger.debug("Could not find period symbol in input: %s", text)
        return None
    else:
        name = text[0:name_end]

    if not name:
        logger.warning("Could not extract name.")
        logger.debug("Could not extract name from input: %s", text)
        return None

    description_start = text.find('"', name_end + 1)
    description_end = text.find('"', description_start + 1)
    if description_end > description_start:
        description = text[description_start:description_end]
    else:
        description = ""

    tags = re.findall(r"#\w+", text.lower())
    if not tags:
        logger.info("Could not extract tags.")
        logger.debug("Could not extract tags from input: %s", text)

    teilchen = TeilchenCreate(
        name=name,
        description=description,
        tags=" ".join(natsorted(tags)),
        text=text,
    )
    logger.debug("Created new Teilchen: %r", teilchen)

    return teilchen


async def load_initial_data() -> Sequence[Teilchen]:
    from .database import engine

    with Session(engine) as session:
        statement = select(
            Teilchen.id, Teilchen.name, Teilchen.description, Teilchen.number, Teilchen.tags
        )  # ty:ignore[no-matching-overload]
        all_teilchen = session.exec(statement).all()
    return all_teilchen