New task operations

Author: Lucas Belo

services/tasks_api/config.py

@@ -0,0 +1,8 @@
+from typing import Optional
+
+from pydantic_settings import BaseSettings
+
+
+class Config(BaseSettings):
+    TABLE_NAME: str = ""
+    DYNAMODB_URL: Optional[str] = None

services/tasks_api/main.py

@@ -1,20 +1,87 @@
-from fastapi import FastAPI
+import uuid
+from typing import Union
+
+import jwt
+from fastapi import Depends, FastAPI, Header
 from fastapi.middleware.cors import CORSMiddleware
 from mangum import Mangum
+from starlette import status
+
+from config import Config
+from models import Task
+from schemas import APITask, APITaskList, CreateTask, CloseTask
+from store import TaskStore
 
 app = FastAPI()
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins="*",
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
+config = Config()
+
+
+def get_task_store() -> TaskStore:
+    return TaskStore(config.TABLE_NAME, dynamodb_url=config.DYNAMODB_URL)
+
+
+def get_user_email(authorization: Union[str, None] = Header(default=None)) -> str:
+    return jwt.decode(authorization, options={"verify_signature": False})[
+        "cognito:username"
+    ]
 
 
 @app.get("/api/health-check/")
 def health_check():
     return {"message": "OK"}
 
 
+@app.post(
+    "/api/create-task", response_model=APITask, status_code=status.HTTP_201_CREATED
+)
+def create_task(
+    parameters: CreateTask,
+    user_email: str = Depends(get_user_email),
+    task_store: TaskStore = Depends(get_task_store),
+):
+    task = Task.create(id_=uuid.uuid4(), title=parameters.title, owner=user_email)
+    task_store.add(task)
+
+    return task
+
+
+@app.get("/api/open-tasks", response_model=APITaskList)
+def open_tasks(
+    user_email: str = Depends(get_user_email),
+    task_store: TaskStore = Depends(get_task_store),
+):
+    results=task_store.list_open(owner=user_email)
+    print("Results: ", results)
+    return APITaskList(results=results)
+
+
 handle = Mangum(app)
+
+
+@app.post("/api/close-task", response_model=APITask)
+def close_task(
+    parameters: CloseTask,
+    user_email: str = Depends(get_user_email),
+    task_store: TaskStore = Depends(get_task_store),
+):
+    task = task_store.get_by_id(task_id=parameters.id, owner=user_email)
+    task.close()
+    task_store.add(task)
+
+    return task
+
+
+@app.get("/api/closed-tasks", response_model=APITaskList)
+def closed_tasks(
+    user_email: str = Depends(get_user_email),
+    task_store: TaskStore = Depends(get_task_store),
+):
+    return APITaskList(results=task_store.list_closed(owner=user_email))

services/tasks_api/models.py

@@ -18,3 +18,7 @@ class Task:
     @classmethod
     def create(cls, id_, title, owner):
         return cls(id_, title, TaskStatus.OPEN, owner)
+    
+
+    def close(self):
+        self.status = TaskStatus.CLOSED

services/tasks_api/poetry.lock

@@ -6,12 +6,15 @@ authors = ["Lucas"]
 readme = "README.md"
 
 [tool.poetry.dependencies]
-python = "^3.11"
+python = "^3.9"
 fastapi = "^0.115.0"
 uvicorn = "^0.31.0"
 httpx = "^0.27.2"
 mangum = "^0.19.0"
 boto3 = "1.21.45"
+pyjwt = "^2.9.0"
+pydantic = "^2.9.2"
+pydantic-settings = "^2.5.2"
 
 
 [tool.poetry.group.dev.dependencies]

services/tasks_api/pyproject.toml

@@ -0,0 +1,28 @@
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict
+
+from models import TaskStatus
+
+
+class CreateTask(BaseModel):
+    title: str
+
+
+class APITask(BaseModel):
+    id: UUID
+    title: str
+    status: TaskStatus
+    owner: str
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class APITaskList(BaseModel):
+    results: list[APITask]
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class CloseTask(BaseModel):
+    id: UUID

services/tasks_api/schemas.py

@@ -6,12 +6,13 @@ useDotenv: true
 
 provider:
   name: aws
-  runtime: python3.11
+  runtime: python3.9
   region: ${opt:region, 'eu-west-1'}
   stage: ${opt:stage, 'development'}
   logRetentionInDays: 30
   environment:
     APP_ENVIRONMENT: ${self:provider.stage}
+    TABLE_NAME: ${self:custom.tableName}
   iam:
     role:
       statements:

services/tasks_api/serverless.yml

@@ -77,4 +77,5 @@ def _list_by_status(self, owner, status):
             last_key = response.get("LastEvaluatedKey")
             if last_key is None:
                 break
+
         return tasks