import json
import os
import uuid
from datetime import datetime
from typing import List, Optional, Dict, Any
from pathlib import Path
class DataModel:
def __init__(self):
self.data_dir = Path(__file__).parent.parent.parent / "data"
self.data_file = self.data_dir / "items.json"
self._ensure_data_directory()
def _ensure_data_directory(self):
"""Ensure the data directory exists"""
self.data_dir.mkdir(exist_ok=True)
def _load_data(self) -> List[Dict[str, Any]]:
"""Load data from JSON file"""
try:
if self.data_file.exists():
with open(self.data_file, 'r', encoding='utf-8') as f:
return json.load(f)
return []
except (json.JSONDecodeError, IOError):
return []
def _save_data(self, data: List[Dict[str, Any]]) -> None:
"""Save data to JSON file"""
with open(self.data_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
async def get_all_items(self) -> List[Dict[str, Any]]:
"""Get all items from the data store"""
return self._load_data()
async def get_item_by_id(self, item_id: str) -> Optional[Dict[str, Any]]:
"""Get a specific item by its ID"""
items = self._load_data()
for item in items:
if item.get('id') == item_id:
return item
return None
async def create_item(self, item_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new item"""
items = self._load_data()
new_item = {
'id': str(uuid.uuid4()),
'name': item_data.get('name', ''),
'description': item_data.get('description', ''),
'category': item_data.get('category', ''),
'createdAt': datetime.utcnow().isoformat() + 'Z',
'updatedAt': datetime.utcnow().isoformat() + 'Z'
}
items.append(new_item)
self._save_data(items)
return new_item
async def update_item(self, item_id: str, update_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Update an existing item by ID"""
items = self._load_data()
for i, item in enumerate(items):
if item.get('id') == item_id:
# Update only provided fields
for key, value in update_data.items():
if key in ['name', 'description', 'category']:
items[i][key] = value
items[i]['updatedAt'] = datetime.utcnow().isoformat() + 'Z'
self._save_data(items)
return items[i]
return None
async def delete_item(self, item_id: str) -> Optional[Dict[str, Any]]:
"""Delete an item by ID"""
items = self._load_data()
for i, item in enumerate(items):
if item.get('id') == item_id:
deleted_item = items.pop(i)
self._save_data(items)
return deleted_item
return None
async def search_items(self, query: str) -> List[Dict[str, Any]]:
"""Search items by query string"""
items = self._load_data()
query_lower = query.lower()
results = []
for item in items:
# Search in name, description, and category
if (query_lower in item.get('name', '').lower() or
query_lower in item.get('description', '').lower() or
query_lower in item.get('category', '').lower()):
results.append(item)
return results
