teilchensammler-cli/src/teilchensammler_cli/models.py

import logging
import uuid

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


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)


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 + 1 : 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(engine) -> Sequence[Teilchen]:
    """Retrieve all Teilchen records from the database.

    Args:
        engine (sqlalchemy.Engine): the engine or connection or whatever

    Returns:
        List of Teilchen, potentially empty
    """
    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()

    logger.debug("Loading initial data: found %s records", len(all_teilchen))
    return all_teilchen


async def add_to_database(tc: TeilchenCreate, engine) -> Teilchen:
    """Add given data as a new record into the database.

    Args:
        engine (sqlalchemy.Engine): the engine or connection or whatever
        tc: Teilchen data (no `id` yet)

    Returns:
        The newly created Teilchen (this time with `id`)
    """
    logger.debug("received: %s", str(tc))
    with Session(engine) as session:
        teilchen = Teilchen.model_validate(tc)
        session.add(teilchen)
        session.commit()
        session.refresh(teilchen)

    logger.debug("created: %s", str(teilchen))
    return teilchen