add feature run_eventql_query

atstaeff · atstaeff · commit 9575a89648b5 · 2025-05-01T21:49:31.000+02:00
diff --git a/eventsourcingdb/client.py b/eventsourcingdb/client.py
@@ -1,7 +1,7 @@
 from collections.abc import AsyncGenerator
 import contextlib
 from types import TracebackType
-from typing import TypeVar
+from typing import Any, TypeVar
 
 from http import HTTPStatus
 import json
@@ -110,9 +110,8 @@ async def write_events(
 
         if response.status_code != HTTPStatus.OK:
             raise ServerError(
-                f'Unexpected response status: '
-                f'{response.status_code} {HTTPStatus(response.status_code).phrase}.'
-            )
+                f'Unexpected response status:  '
+        )
 
         response_data = await response.body.read()
         response_data = bytes.decode(response_data, encoding='utf-8')
@@ -144,9 +143,8 @@ async def read_events(
         async with response:
             if response.status_code != HTTPStatus.OK:
                 raise ServerError(
-                    f'Unexpected response status: '
-                    f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
-                )
+                    f'Unexpected response status: {response}'
+            )
             async for raw_message in response.body:
                 message = parse_raw_message(raw_message)
 
@@ -189,11 +187,36 @@ async def read_events(
                     f'{message}.'
                 )
 
-    async def run_eventql_query(self, query: str) -> AsyncGenerator[Event]:
-        # TODO: read events nehmen. Das Responsehandling ist gleich wie
-        # read_events. ein object was eine query property hat, Return ist ein any.
-        # da kann alles mögliche sein.
-        raise NotImplementedError("run_eventql_query is not implemented yet.")
+    async def run_eventql_query(self, query: str) -> AsyncGenerator[Any, None]:
+        request_body = json.dumps({
+            'query': query,
+        })
+        response: Response = await self.__http_client.post(
+            path='/api/v1/run-eventql-query',
+            request_body=request_body,
+        )
+
+        async with response:
+            if response.status_code != HTTPStatus.OK:
+                raise ServerError(
+                    f'Unexpected response status: {response}'
+                )
+            async for raw_message in response.body:
+                message = parse_raw_message(raw_message)
+
+                if is_stream_error(message):
+                    raise ServerError(f'{message["payload"]["error"]}.')
+
+                if message.get('type') == 'row':
+                    payload = message['payload']
+                
+                    yield payload
+                    continue
+
+                raise ServerError(
+                    f'Failed to execute EventQL query, an unexpected stream item was received: '
+                    f'{message}.'
+                )
 
     async def observe_events(
         self,
@@ -205,17 +228,15 @@ async def observe_events(
             'options': options.to_json()
         })
 
-        response: Response
-        response = await self.http_client.post(
+        response : Response = await self.http_client.post(
             path='/api/v1/observe-events',
             request_body=request_body,
         )
 
-        with response:
+        async with response:
             if response.status_code != HTTPStatus.OK:
                 raise ServerError(
-                    f'Unexpected response status: '
-                    f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
+                    f'Unexpected response status: {response}'
                 )
             async for raw_message in response.body:
                 message = parse_raw_message(raw_message)
@@ -256,23 +277,17 @@ async def register_event_schema(self, event_type: str, json_schema: dict) -> Non
             'schema': json_schema,
         })
 
-        response: Response
-        response = await self.http_client.post(
+        response: Response = await self.http_client.post(
             path='/api/v1/register-event-schema',
             request_body=request_body,
         )
 
-        with response:
+        async with response:
             if response.status_code != HTTPStatus.OK:
                 raise ServerError(
-                    'Unexpected response status: '
-                    f'{response.status_code} '
-                    f'{HTTPStatus(response.status_code).phrase} '
-                    f'{str(response)} '
+                    f'Unexpected response status: {response} '
                 )
 
-        return
-
     async def read_subjects(
         self,
         base_subject: str
@@ -281,19 +296,15 @@ async def read_subjects(
             'baseSubject': base_subject
         })
 
-        response: Response
-        response = await self.http_client.post(
+        response : Response = await self.http_client.post(
             path='/api/v1/read-subjects',
             request_body=request_body,
         )
 
-        with response:
+        async with response:
             if response.status_code != HTTPStatus.OK:
                 raise ServerError(
-                    'Unexpected response status: '
-                    f'{response.status_code} '
-                    f'{HTTPStatus(response.status_code).phrase} '
-                    f'{str(response)} '
+                    f'Unexpected response status: {response}'
                 )
             async for raw_message in response.body:
                 message = parse_raw_message(raw_message)
@@ -322,11 +333,10 @@ async def read_event_types(self) -> AsyncGenerator[EventType]:
         except Exception as other_error:
             raise InternalError(str(other_error)) from other_error
 
-        with response:
+        async with response:
             if response.status_code != HTTPStatus.OK:
                 raise ServerError(
-                    'Unexpected response status: '
-                    f'{response.status_code} {HTTPStatus(response.status_code).phrase}'
+                    f'Unexpected response status: {response}'
                 )
             async for raw_message in response.body:
                 message = parse_raw_message(raw_message)
diff --git a/eventsourcingdb/http_client/http_client.py b/eventsourcingdb/http_client/http_client.py
@@ -45,21 +45,19 @@ async def post(self, path: str, request_body: str) -> Response:
         if self.__session is None:
             raise CustomError()
 
-        async def __request_executor() -> Response:
-            url_path = url.join_segments(self.__base_url, path)
-            headers = get_post_headers(self.__api_token)
+        url_path = url.join_segments(self.__base_url, path)
+        headers = get_post_headers(self.__api_token)
 
-            async_response = await self.__session.post(  # type: ignore
-                url_path,
-                data=request_body,
-                headers=headers,
-            )
+        async_response = await self.__session.post(  # type: ignore
+            url_path,
+            data=request_body,
+            headers=headers,
+        )
 
-            response = Response(async_response)
+        response = Response(async_response)
 
-            return response
+        return response
 
-        return await __request_executor()
 
     async def get(
         self,
diff --git a/eventsourcingdb/http_client/response.py b/eventsourcingdb/http_client/response.py
@@ -39,4 +39,8 @@ def body(self) -> StreamReader:
         return self.__response.content
 
     def __str__(self) -> str:
-        return f'Response(status_code={self.status_code}, headers={dict(self.headers)})'
+        status_code_text = f'{self.status_code} {HTTPStatus(self.status_code).phrase}'
+        header_text = dict(self.headers)
+        body_text = self.body.read_nowait().decode("utf-8")
+        status_code_text = f'{self.status_code} {HTTPStatus(self.status_code).phrase}'
+        return f'status_code={status_code_text}, headers={header_text}, body={body_text}'
diff --git a/tests/test_register_event_schema.py b/tests/test_register_event_schema.py
@@ -44,7 +44,7 @@ async def test_throws_error_if_schema_conflicts_with_existing_events(
             ]
         )
 
-        with pytest.raises(ServerError, match="Bad Request"):
+        with pytest.raises(ServerError, match='missing properties'):
             await client.register_event_schema(
                 "com.gornisht.ekht",
                 {
@@ -69,7 +69,7 @@ async def test_throws_error_if_schema_already_exists(
             }
         )
 
-        with pytest.raises(ServerError):
+        with pytest.raises(ServerError, match='schema already exists'):
             await client.register_event_schema(
                 "com.gornisht.ekht",
                 {
@@ -86,7 +86,7 @@ async def test_throws_error_if_schema_is_invalid(
     ):
         client = database.get_client()
 
-        with pytest.raises(ServerError):
+        with pytest.raises(ServerError,  match='value must be "object"'):
             await client.register_event_schema(
                 "com.gornisht.ekht",
                 {
diff --git a/tests/test_run_eventql_query.py b/tests/test_run_eventql_query.py
@@ -0,0 +1,80 @@
+import asyncio
+
+from aiohttp import ClientConnectorDNSError
+import pytest
+
+from eventsourcingdb.errors.server_error import ServerError
+from eventsourcingdb.event.event_candidate import EventCandidate
+
+from .conftest import TestData
+from .shared.database import Database
+from .shared.event.assert_event import assert_event_equals
+
+
+class TestRunEventQLQuery:
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_throws_error_if_server_is_not_reachable(
+        database: Database
+    ):
+        client = database.get_client("with_invalid_url")
+
+        with pytest.raises(ClientConnectorDNSError):
+            async for _ in client.run_eventql_query('FROM e IN events PROJECT INTO e'):
+                pass
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_reads_no_rows_if_query_does_not_return_any_rows(
+        database: Database
+    ):
+        client = database.get_client()
+        
+        did_read_rows = False
+        async for _ in client.run_eventql_query('FROM e IN events PROJECT INTO e'):
+            did_read_rows = True
+        
+        assert did_read_rows is False
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_reads_all_rows_the_query_returns(
+        database: Database
+    ):
+        client = database.get_client()
+        
+        first_event = EventCandidate(
+            source='https://www.eventsourcingdb.io',
+            subject='/test',
+            type='io.eventsourcingdb.test',
+            data={
+                'value': 23,
+            },
+        )
+        
+        second_event = EventCandidate(
+            source='https://www.eventsourcingdb.io',
+            subject='/test',
+            type='io.eventsourcingdb.test',
+            data={
+                'value': 42,
+            },
+        )
+        
+        await client.write_events([first_event, second_event])
+        
+        rows_read = []
+        async for row in client.run_eventql_query('FROM e IN events PROJECT INTO e'):
+            rows_read.append(row)
+        
+        assert len(rows_read) == 2
+        
+        first_row = rows_read[0]
+        # Use dictionary access instead of attribute access
+        assert first_row['id'] == '0'
+        assert first_row['data']['value'] == 23
+        
+        second_row = rows_read[1]
+        assert second_row['id'] == '1'
+        assert second_row['data']['value'] == 42
+

Original file line number	Diff line number	Diff line change
`@@ -44,7 +44,7 @@ async def test_throws_error_if_schema_conflicts_with_existing_events(`
`44`	`44`	`]`
`45`	`45`	`)`
`46`	`46`
`47`		`- with pytest.raises(ServerError, match="Bad Request"):`
	`47`	`+ with pytest.raises(ServerError, match='missing properties'):`
`48`	`48`	`await client.register_event_schema(`
`49`	`49`	`"com.gornisht.ekht",`
`50`	`50`	`{`
`@@ -69,7 +69,7 @@ async def test_throws_error_if_schema_already_exists(`
`69`	`69`	`}`
`70`	`70`	`)`
`71`	`71`
`72`		`- with pytest.raises(ServerError):`
	`72`	`+ with pytest.raises(ServerError, match='schema already exists'):`
`73`	`73`	`await client.register_event_schema(`
`74`	`74`	`"com.gornisht.ekht",`
`75`	`75`	`{`
`@@ -86,7 +86,7 @@ async def test_throws_error_if_schema_is_invalid(`
`86`	`86`	`):`
`87`	`87`	`client = database.get_client()`
`88`	`88`
`89`		`- with pytest.raises(ServerError):`
	`89`	`+ with pytest.raises(ServerError, match='value must be "object"'):`
`90`	`90`	`await client.register_event_schema(`
`91`	`91`	`"com.gornisht.ekht",`
`92`	`92`	`{`