Execute a GQL query Args: session_id: Session ID from create_session() query: GQL query string Returns: QueryResult with rows and metadata Raises: GraphLiteError: If query execution fails
(self, session_id: str, query: str)
| 256 | return session_id |
| 257 | |
| 258 | def query(self, session_id: str, query: str) -> QueryResult: |
| 259 | """ |
| 260 | Execute a GQL query |
| 261 | |
| 262 | Args: |
| 263 | session_id: Session ID from create_session() |
| 264 | query: GQL query string |
| 265 | |
| 266 | Returns: |
| 267 | QueryResult with rows and metadata |
| 268 | |
| 269 | Raises: |
| 270 | GraphLiteError: If query execution fails |
| 271 | """ |
| 272 | if not self._db: |
| 273 | raise GraphLiteError(ErrorCode.NULL_POINTER, "Database is closed") |
| 274 | |
| 275 | error = ctypes.c_int(0) |
| 276 | result_ptr = _lib.graphlite_query( |
| 277 | self._db, |
| 278 | session_id.encode('utf-8'), |
| 279 | query.encode('utf-8'), |
| 280 | ctypes.byref(error) |
| 281 | ) |
| 282 | |
| 283 | if not result_ptr: |
| 284 | raise GraphLiteError( |
| 285 | ErrorCode(error.value), |
| 286 | f"Query failed: {query[:100]}" |
| 287 | ) |
| 288 | |
| 289 | try: |
| 290 | # Copy the string before freeing |
| 291 | result_json = ctypes.string_at(result_ptr).decode('utf-8') |
| 292 | result_data = json.loads(result_json) |
| 293 | return QueryResult(result_data) |
| 294 | except json.JSONDecodeError as e: |
| 295 | raise GraphLiteError(ErrorCode.JSON_ERROR, f"Invalid JSON response: {e}") |
| 296 | finally: |
| 297 | _lib.graphlite_free_string(result_ptr) |
| 298 | |
| 299 | def execute(self, session_id: str, statement: str) -> None: |
| 300 | """ |
no test coverage detected