Create a new cube. Args: cube_name (str): Name of the cube. owner_id (str): ID of the cube owner. cube_path (str, optional): Path to the cube. cube_id (str, optional): Custom cube ID. If None, generates UUID. Returns: str:
(
self,
cube_name: str,
owner_id: str,
cube_path: str | None = None,
cube_id: str | None = None,
)
| 239 | session.close() |
| 240 | |
| 241 | def create_cube( |
| 242 | self, |
| 243 | cube_name: str, |
| 244 | owner_id: str, |
| 245 | cube_path: str | None = None, |
| 246 | cube_id: str | None = None, |
| 247 | ) -> str: |
| 248 | """Create a new cube. |
| 249 | |
| 250 | Args: |
| 251 | cube_name (str): Name of the cube. |
| 252 | owner_id (str): ID of the cube owner. |
| 253 | cube_path (str, optional): Path to the cube. |
| 254 | cube_id (str, optional): Custom cube ID. If None, generates UUID. |
| 255 | |
| 256 | Returns: |
| 257 | str: The created cube ID. |
| 258 | |
| 259 | Raises: |
| 260 | ValueError: If owner doesn't exist. |
| 261 | """ |
| 262 | session = self._get_session() |
| 263 | try: |
| 264 | # Validate owner exists |
| 265 | owner = session.query(User).filter(User.user_id == owner_id).first() |
| 266 | if not owner: |
| 267 | raise ValueError(f"User with ID '{owner_id}' does not exist") |
| 268 | |
| 269 | cube = Cube( |
| 270 | cube_name=cube_name, |
| 271 | owner_id=owner_id, |
| 272 | cube_path=cube_path, |
| 273 | cube_id=cube_id or str(uuid.uuid4()), |
| 274 | ) |
| 275 | session.add(cube) |
| 276 | |
| 277 | # Add owner to cube users |
| 278 | cube.users.append(owner) |
| 279 | |
| 280 | session.commit() |
| 281 | logger.info(f"Cube '{cube_name}' created with ID: {cube.cube_id}") |
| 282 | return cube.cube_id |
| 283 | except Exception as e: |
| 284 | session.rollback() |
| 285 | logger.error(f"Error creating cube: {e}") |
| 286 | raise |
| 287 | finally: |
| 288 | session.close() |
| 289 | |
| 290 | def get_cube(self, cube_id: str) -> Cube | None: |
| 291 | """Get cube by ID. |
no test coverage detected