Spaces:
Sleeping
Sleeping
File size: 2,040 Bytes
cdec65b | 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 | import os
import uuid
from typing import List, Dict, Optional
from datetime import datetime
from pymongo import MongoClient
class SpacesManager:
"""Manages creation and retrieval of Spaces using MongoDB Atlas."""
def __init__(self):
self.mongo_uri = os.getenv("MONGO_URI")
if self.mongo_uri:
self.client = MongoClient(self.mongo_uri)
self.db = self.client["notebookpro_db"]
self.collection = self.db["spaces"]
else:
print("WARNING: MONGO_URI not set. SpacesManager will fail.")
self.collection = None
def get_all_spaces(self) -> List[Dict]:
"""Retrieve all spaces from MongoDB."""
if self.collection is None: return []
# Return all documents, but exclude the internal MongoDB '_id' field
return list(self.collection.find({}, {"_id": 0}))
def get_space(self, space_id: str) -> Optional[Dict]:
"""Retrieve a single space by ID."""
if self.collection is None: return None
return self.collection.find_one({"id": space_id}, {"_id": 0})
def create_space(self, name: str) -> Dict:
"""Create a new space and save it to MongoDB."""
# Generate a clean ID from the name
base_id = "".join(c for c in name.lower() if c.isalnum() or c == ' ').replace(' ', '_')
space_id = base_id
# Ensure ID is unique
while self.collection.find_one({"id": space_id}):
space_id = f"{base_id}_{uuid.uuid4().hex[:4]}"
space_data = {
"id": space_id,
"name": name,
"created_at": datetime.now().isoformat()
}
self.collection.insert_one(space_data)
# Remove the injected _id before returning to FastAPI
space_data.pop('_id', None)
return space_data
def delete_space(self, space_id: str):
"""Delete a space from MongoDB."""
if self.collection is not None:
self.collection.delete_one({"id": space_id}) |