Create a new session for the given user Args: username: Username for the session Returns: Session ID string Raises: GraphLiteError: If session creation fails
(self, username: str)
| 220 | ) |
| 221 | |
| 222 | def create_session(self, username: str) -> str: |
| 223 | """ |
| 224 | Create a new session for the given user |
| 225 | |
| 226 | Args: |
| 227 | username: Username for the session |
| 228 | |
| 229 | Returns: |
| 230 | Session ID string |
| 231 | |
| 232 | Raises: |
| 233 | GraphLiteError: If session creation fails |
| 234 | """ |
| 235 | if not self._db: |
| 236 | raise GraphLiteError(ErrorCode.NULL_POINTER, "Database is closed") |
| 237 | |
| 238 | error = ctypes.c_int(0) |
| 239 | session_id_ptr = _lib.graphlite_create_session( |
| 240 | self._db, |
| 241 | username.encode('utf-8'), |
| 242 | ctypes.byref(error) |
| 243 | ) |
| 244 | |
| 245 | if not session_id_ptr: |
| 246 | raise GraphLiteError( |
| 247 | ErrorCode(error.value), |
| 248 | f"Failed to create session for user '{username}'" |
| 249 | ) |
| 250 | |
| 251 | # Copy the string before freeing |
| 252 | session_id = ctypes.string_at(session_id_ptr).decode('utf-8') |
| 253 | _lib.graphlite_free_string(session_id_ptr) |
| 254 | self._sessions.add(session_id) |
| 255 | |
| 256 | return session_id |
| 257 | |
| 258 | def query(self, session_id: str, query: str) -> QueryResult: |
| 259 | """ |
no test coverage detected