Spaces:
Sleeping
Sleeping
File size: 2,390 Bytes
08af9fd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """
In-memory storage implementation for tasks.
All data is stored in memory only (no persistent storage).
"""
from typing import Dict, Optional
from src.models.task import Task
class InMemoryStorage:
"""
In-memory storage for tasks using a dictionary with ID as key.
Provides O(1) lookup by ID and maintains uniqueness of task IDs.
"""
def __init__(self):
self._tasks: Dict[int, Task] = {}
self._next_id = 1
def add_task(self, task: Task) -> Task:
"""Insert a new task with auto-generated unique ID."""
# If the task doesn't have an ID yet, assign one
if task.id is None or task.id == 0:
task.id = self._next_id
self._next_id += 1
# Ensure ID is unique
if task.id in self._tasks:
raise ValueError(f"Task with ID {task.id} already exists")
self._tasks[task.id] = task
return task
def get_task(self, task_id: int) -> Optional[Task]:
"""Retrieve a task by its ID."""
return self._tasks.get(task_id)
def update_task(self, task_id: int, updated_task: Task) -> Optional[Task]:
"""Update existing task attributes."""
if task_id not in self._tasks:
return None
# Preserve the original ID
updated_task.id = task_id
self._tasks[task_id] = updated_task
return updated_task
def delete_task(self, task_id: int) -> bool:
"""Remove a task by its ID."""
if task_id not in self._tasks:
return False
del self._tasks[task_id]
return True
def list_tasks(self) -> list:
"""Retrieve all tasks in the collection."""
return list(self._tasks.values())
def find_tasks(self, **criteria) -> list:
"""Search/filter tasks based on criteria (e.g., completed status)."""
tasks = self.list_tasks()
if 'completed' in criteria:
completed = criteria['completed']
tasks = [task for task in tasks if task.completed == completed]
if 'title' in criteria:
title = criteria['title']
tasks = [task for task in tasks if title.lower() in task.title.lower()]
return tasks
def get_next_id(self) -> int:
"""Allocate and return the next unique ID."""
next_id = self._next_id
self._next_id += 1
return next_id |