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: import re from natsort import natsorted if not text: logger.error("Empty text.") return None name_end = text.find(".") name = text[0:name_end] if not name: logger.error("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 ) all_teilchen = session.exec(statement).all() return all_teilchen