initial commit

Author: wjohnsto

app/components/todos/__init__.py

@@ -0,0 +1,50 @@
+from typing import AsyncIterator, Never
+from fastapi import APIRouter, FastAPI
+from fastapi.concurrency import asynccontextmanager
+from pydantic import BaseModel
+
+from app.redis import redis
+from app.components.todos.store import Todo, TodoDocument, TodoStatus, TodoStore, Todos
+
+todos = TodoStore(redis)
+
+@asynccontextmanager
+async def lifespan(_: FastAPI) -> AsyncIterator[Never]:
+    # before
+    await todos.initialize()
+    yield  # type: ignore
+    # after
+    return
+
+router = APIRouter(lifespan=lifespan)
+
+@router.get("/", tags=["todos"])
+async def all() -> Todos:
+    return await todos.all()
+
+@router.get("/search", tags=["todos"])
+async def search(name: str | None = None, status: TodoStatus | None = None) -> Todos:
+    return await todos.search(name, status)
+
+@router.get("/{id}", tags=["todos"])
+async def one(id: str) -> Todo:
+    return await todos.one(id)
+
+class CreateTodo(BaseModel):
+    id: str | None = None
+    name: str
+
+@router.post("/", tags=["todos"])
+async def create(todo: CreateTodo) -> TodoDocument:
+    return await todos.create(todo.id, todo.name)
+
+class UpdateTodo(BaseModel):
+    status: TodoStatus
+
+@router.patch("/{id}", tags=["todos"])
+async def update(id: str, todo: UpdateTodo) -> Todo:
+    return await todos.update(id, todo.status)
+
+@router.delete("/{id}", tags=["todos"])
+async def delete(id: str) -> None:
+    return await todos.delete(id)

app/components/todos/router.py

@@ -0,0 +1,238 @@
+import re
+from datetime import datetime
+from enum import Enum
+from typing import Any, List, Optional
+from uuid import uuid4
+
+from pydantic import BaseModel, Field, TypeAdapter, field_serializer
+from pydantic_core import from_json
+from redis.asyncio import Redis
+from redis.commands.search.document import Document
+from redis.commands.search.field import TextField
+from redis.commands.search.query import Query
+from redis.commands.search.indexDefinition import IndexDefinition, IndexType
+from redis.exceptions import ResponseError
+
+from app.logger import logger
+
+TODOS_INDEX = "todos-idx"
+TODOS_PREFIX = "todos:"
+
+
+class TodoStatus(str, Enum):
+    todo = 'todo'
+    in_progress = 'in progress'
+    complete = 'complete'
+
+
+class Todo(BaseModel):
+    name: str
+    status: TodoStatus
+    created_date: datetime = None
+    updated_date: datetime = None
+
+    @field_serializer('created_date')
+    def serialize_created_date(self, dt: datetime, _info):
+        return dt.strftime('%Y-%m-%dT%H:%M:%SZ')
+
+    @field_serializer('updated_date')
+    def serialize_updated_date(self, dt: datetime, _info):
+        return dt.strftime('%Y-%m-%dT%H:%M:%SZ')
+
+class TodoDocument(BaseModel):
+    id: str
+    value: Todo
+
+    # def __init__(self, doc: Any, test: Any):
+    #     print("test")
+    #     print(doc)
+    #     super().__init__(id=doc.id, document=Todo(**from_json(doc.json, allow_partial=True)))
+
+class Todos(BaseModel):
+    total: int
+    documents: List[TodoDocument]
+
+class TodoStore:
+    """Stores todos"""
+
+    def __init__(self, redis: Redis):
+        self.redis = redis
+        self.INDEX = TODOS_INDEX
+        self.PREFIX = TODOS_PREFIX
+
+    async def initialize(self) -> None:
+        await self.create_index_if_not_exists()
+        return None
+
+    async def have_index(self) -> bool:
+        try:
+            info = await self.redis.ft(self.INDEX).info()  # type: ignore
+        except ResponseError as e:
+            if "Unknown index name" in str(e):
+                logger.info(f'Index {self.INDEX} does not exist')
+                return False
+
+        logger.info(f"Index {self.INDEX} already exists")
+        return True
+
+    async def create_index_if_not_exists(self) -> None:
+        if await self.have_index():
+            return None
+
+        logger.debug(f"Creating index {self.INDEX}")
+
+        schema = (
+            TextField("$.name", as_name="name"),
+            TextField("$.status", as_name="status"),
+        )
+
+        try:
+            await self.redis.ft(self.INDEX).create_index(  # type: ignore
+                schema,
+                definition=IndexDefinition(  # type: ignore
+                    prefix=[TODOS_PREFIX], index_type=IndexType.JSON
+                ),
+            )
+        except Exception as e:
+            logger.error(f"Error setting up index {self.INDEX}: {e}")
+            raise
+
+        logger.debug(f"Index {self.INDEX} created successfully")
+
+        return None
+
+    async def drop_index(self) -> None:
+        if not await self.have_index():
+            return None
+        
+        try:
+            await self.redis.ft(self.INDEX).dropindex()
+        except Exception as e:
+            logger.error(f"Error dropping index ${self.INDEX}: {e}")
+            raise
+
+        logger.debug(f"Index {self.INDEX} dropped successfully")
+
+        return None
+    
+    def format_id(self, id: str) -> str:
+        if re.match(f'^{self.PREFIX}', id):
+            return id
+        
+        return f'{self.PREFIX}{id}'
+    
+    def parse_todo_document(self, todo: Document) -> TodoDocument:
+        return TodoDocument(id=todo.id, value=Todo(**from_json(todo.json, allow_partial=True)))
+
+    def parse_todo_documents(self, todos: list[Document]) -> Todos:
+        todo_docs = [];
+
+        for doc in todos:
+            todo_docs.append(self.parse_todo_document(doc))
+
+        return todo_docs
+
+    async def all(self):
+        try:
+            result = await self.redis.ft(self.INDEX).search("*")
+            return Todos(total=result.total, documents=self.parse_todo_documents(result.docs))
+        except Exception as e:
+            logger.error(f"Error getting all todos: {e}")
+            raise
+    
+    async def one(self, id: str) -> Todo:
+        id = self.format_id(id)
+
+        try:
+            json = await self.redis.json().get(id)
+        except Exception as e:
+            logger.error(f"Error getting todo ${id}: {e}")
+            raise
+
+        return Todo(**json)
+    
+    async def search(self, name: str | None, status: TodoStatus | None) -> Todo:
+        searches = []
+
+        if name is not None and len(name) > 0:
+            searches.append(f'@name:({name})')
+        
+        if status is not None and len(status) > 0:
+            searches.append(f'@status:{status.value}')
+
+        try:
+            result = await self.redis.ft(self.INDEX).search(Query(' '.join(searches)))
+            return Todos(total=result.total, documents=self.parse_todo_documents(result.docs))
+        except Exception as e:
+            logger.error(f"Error getting todo {id}: {e}")
+            raise
+    
+    async def create(self, id: Optional[str], name: Optional[str]) -> TodoDocument:
+        dt = datetime.now()
+
+        if name is None:
+            raise Exception("Todo must have a name")
+        
+        if id is None:
+            id = str(uuid4())
+        
+        todo = TodoDocument(**{
+            "id": self.format_id(id),
+            "value": {
+                "name": name,
+                "status": "todo",
+                "created_date": dt,
+                "updated_date": dt,
+            }
+        })
+
+        try:
+            result = await self.redis.json().set(todo.id, "$", todo.value.model_dump())
+        except Exception as e:
+            logger.error(f'Error creating todo {todo}: {e}')
+            raise
+
+        if result != True:
+            raise Exception(f'Error creating todo {todo}')
+        
+        return todo
+
+    async def update(self, id: str, status: str) -> Todo:
+        dt = datetime.now()
+
+        todo = await self.one(id)
+
+        todo.status = status
+        todo.updated_date = dt
+
+        try:
+            result = await self.redis.json().set(self.format_id(id), "$", todo.model_dump())
+        except Exception as e:
+            logger.error(f'Error updating todo {todo}: {e}')
+            raise
+
+        if result != True:
+            raise Exception(f'Error creating todo {todo}')
+            
+        return todo
+
+    async def delete(self, id: str) -> None:
+        try:
+            await self.redis.json().delete(self.format_id(id))
+        except Exception as e:
+            logger.error(f'Error deleting todo {id}: {e}')
+            raise
+
+        return None
+    
+    async def delete_all(self) -> None:
+        todos = await self.all()
+
+        try:
+            for todo in todos.documents:
+                await self.redis.json().delete(todo.id)
+        except Exception as e:
+            logger.error(f'Error deleting todos: {e}')
+            raise
+        
+        return None

app/components/todos/store.py

@@ -3,35 +3,12 @@
 
 import os
 from contextlib import asynccontextmanager
-from typing import Any, AsyncIterator, Never, Union
 
 from fastapi import FastAPI
 
 import app.redis as redis
-from app.todos import Todos
+from app.components.todos.router import router as todos_router
 
-redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379")
-client = redis.get_client(redis_url)
-todos = Todos(client)
 
-
-@asynccontextmanager
-async def lifespan(_: FastAPI) -> AsyncIterator[Never]:
-    # before
-    await todos.initialize()
-    yield  # type: ignore
-    # after
-    return
-
-
-app = FastAPI(lifespan=lifespan)
-
-
-@app.get("/")
-async def read_root() -> bool:
-    return await todos.have_index()
-
-
-@app.get("/items/{item_id}")
-def read_item(item_id: int, q: Union[str, None] = None) -> Any:
-    return {"item_id": item_id, "q": q}
+app = FastAPI()
+app.include_router(router=todos_router, prefix="/api/todos")

app/main.py

@@ -1,3 +1,4 @@
+import os
 from typing import Optional
 
 from redis.asyncio import Redis
@@ -7,20 +8,11 @@
 
 from app.logger import logger
 
-client: Optional[Redis] = None
-
-
-def get_client(url: str = "redis://localhost:6379") -> Redis:
-    global client
-
-    if client is None:
-        logger.info(f"Creating redis client for url: {url}")
-        client = Redis.from_url(
-            url,
-            decode_responses=True,
-            retry=Retry(ExponentialBackoff(cap=10, base=1), 25),
-            retry_on_error=[ConnectionError, TimeoutError, ConnectionResetError],
-            health_check_interval=1,
-        )
-
-    return client
+redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379")
+redis = Redis.from_url(
+    redis_url,
+    decode_responses=True,
+    retry=Retry(ExponentialBackoff(cap=10, base=1), 25),
+    retry_on_error=[ConnectionError, TimeoutError, ConnectionResetError],
+    health_check_interval=1,
+)