This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 8d454cb075 feat: modernize Python Client syntax (PEP 585, PEP 604)
(#4628)
8d454cb075 is described below
commit 8d454cb075f06be4e37dac344b46f563af7703de
Author: airwish <[email protected]>
AuthorDate: Fri Jun 26 23:22:50 2026 +0800
feat: modernize Python Client syntax (PEP 585, PEP 604) (#4628)
---
.../streampipes/client/client.py | 5 +-
.../streampipes/client/config.py | 7 ++-
.../streampipes/client/credential_provider.py | 13 +++--
.../streampipes/endpoint/api/data_lake_measure.py | 28 +++++-----
.../streampipes/endpoint/api/data_stream.py | 5 +-
.../streampipes/endpoint/api/version.py | 7 ++-
.../streampipes/endpoint/endpoint.py | 11 ++--
.../streampipes/function_zoo/river_function.py | 15 +++---
.../streampipes/functions/broker/consumer.py | 2 +-
.../functions/broker/kafka/kafka_consumer.py | 2 +-
.../functions/broker/kafka/kafka_publisher.py | 4 +-
.../functions/broker/nats/nats_consumer.py | 2 +-
.../functions/broker/nats/nats_publisher.py | 4 +-
.../functions/broker/output_collector.py | 5 +-
.../streampipes/functions/broker/publisher.py | 4 +-
.../streampipes/functions/function_handler.py | 10 ++--
.../streampipes/functions/registration.py | 5 +-
.../streampipes/functions/streampipes_function.py | 10 ++--
.../functions/utils/async_iter_handler.py | 7 +--
.../functions/utils/data_stream_context.py | 3 +-
.../functions/utils/data_stream_generator.py | 5 +-
.../functions/utils/function_context.py | 3 +-
.../streampipes/model/common.py | 59 +++++++++++-----------
.../model/container/data_lake_measures.py | 3 +-
.../streampipes/model/container/data_streams.py | 3 +-
.../model/container/resource_container.py | 7 ++-
.../streampipes/model/container/versions.py | 3 +-
.../model/resource/data_lake_measure.py | 11 ++--
.../streampipes/model/resource/data_series.py | 10 ++--
.../streampipes/model/resource/data_stream.py | 31 ++++++------
.../streampipes/model/resource/exceptions.py | 3 +-
.../model/resource/function_definition.py | 11 ++--
.../streampipes/model/resource/query_result.py | 12 ++---
.../streampipes/model/resource/resource.py | 3 +-
.../streampipes/model/resource/version.py | 5 +-
.../tests/client/test_endpoint.py | 3 +-
36 files changed, 152 insertions(+), 169 deletions(-)
diff --git a/streampipes-client-python/streampipes/client/client.py
b/streampipes-client-python/streampipes/client/client.py
index 11f97348cc..2b8faff3a6 100644
--- a/streampipes-client-python/streampipes/client/client.py
+++ b/streampipes-client-python/streampipes/client/client.py
@@ -25,7 +25,6 @@ from __future__ import annotations
import logging
import sys
-from typing import Dict, Optional
from requests import Session
@@ -120,7 +119,7 @@ class StreamPipesClient:
def __init__(
self,
client_config: StreamPipesClientConfig,
- logging_level: Optional[int] = logging.INFO,
+ logging_level: int | None = logging.INFO,
):
# validate client config
# `https_disabled` and `port` 443 is an invalid configuration
@@ -228,7 +227,7 @@ class StreamPipesClient:
return cls(client_config=client_config, logging_level=logging_level)
@property
- def http_headers(self) -> Dict[str, str]:
+ def http_headers(self) -> dict[str, str]:
"""Returns the HTTP headers used for all requests.
The HTTP headers are composed of the authentication headers supplied
by the credential
diff --git a/streampipes-client-python/streampipes/client/config.py
b/streampipes-client-python/streampipes/client/config.py
index 36b0618ad9..46a7ecf936 100644
--- a/streampipes-client-python/streampipes/client/config.py
+++ b/streampipes-client-python/streampipes/client/config.py
@@ -21,7 +21,6 @@ Configuration class for the StreamPipes client.
from dataclasses import dataclass, field
-from typing import Dict, Optional
__all__ = [
"StreamPipesClientConfig",
@@ -61,6 +60,6 @@ class StreamPipesClientConfig:
credential_provider: CredentialProvider
host_address: str
- https_disabled: Optional[bool] = False
- port: Optional[int] = 80
- additional_headers: Optional[Dict[str, str]] = field(default_factory=dict)
+ https_disabled: bool | None = False
+ port: int | None = 80
+ additional_headers: dict[str, str] | None = field(default_factory=dict)
diff --git
a/streampipes-client-python/streampipes/client/credential_provider.py
b/streampipes-client-python/streampipes/client/credential_provider.py
index edac71dde8..eea5122581 100644
--- a/streampipes-client-python/streampipes/client/credential_provider.py
+++ b/streampipes-client-python/streampipes/client/credential_provider.py
@@ -25,7 +25,6 @@ from __future__ import annotations
import os
from abc import ABC, abstractmethod
-from typing import Dict, Optional
__all__ = [
"CredentialProvider",
@@ -41,7 +40,7 @@ class CredentialProvider(ABC):
Must be inherited by all credential providers.
"""
- def make_headers(self, http_headers: Optional[Dict[str, str]] = None) ->
Dict[str, str]:
+ def make_headers(self, http_headers: dict[str, str] | None = None) ->
dict[str, str]:
"""Creates the HTTP headers for the specific credential provider.
Concrete authentication headers must be defined in the implementation
of a credential provider.
@@ -68,7 +67,7 @@ class CredentialProvider(ABC):
@property
@abstractmethod
- def _authentication_headers(self) -> Dict[str, str]:
+ def _authentication_headers(self) -> dict[str, str]:
"""Provides the HTTP headers used for the authentication with the
concrete `CredentialProvider`.
Returns
@@ -148,8 +147,8 @@ class StreamPipesApiKeyCredentials(CredentialProvider):
def __init__(
self,
- username: Optional[str] = None,
- api_key: Optional[str] = None,
+ username: str | None = None,
+ api_key: str | None = None,
):
# if both parameters are passed we can add them directly to the
instance
if all({username, api_key}):
@@ -196,7 +195,7 @@ class StreamPipesApiKeyCredentials(CredentialProvider):
)
@property
- def _authentication_headers(self) -> Dict[str, str]:
+ def _authentication_headers(self) -> dict[str, str]:
"""Provides the HTTP headers used for the authentication with the API
token.
Returns
@@ -238,7 +237,7 @@ class StreamPipesTokenCredentials(CredentialProvider):
self.jwt = jwt
@property
- def _authentication_headers(self) -> Dict[str, str]:
+ def _authentication_headers(self) -> dict[str, str]:
"""Provides the HTTP headers used for authentication with the JWT
token.
Returns
diff --git
a/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py
b/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py
index 11d042ad93..5848ec620e 100644
--- a/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py
+++ b/streampipes-client-python/streampipes/endpoint/api/data_lake_measure.py
@@ -22,7 +22,7 @@ This endpoint allows to consume data stored in StreamPipes'
data lake.
from datetime import datetime
from json import dumps
from math import ceil
-from typing import Any, Dict, List, Literal, Optional, Tuple, Type
+from typing import Any, Literal
from pandas import DataFrame
from pydantic import BaseModel, ConfigDict, Field, StrictInt, ValidationError,
field_validator
@@ -81,17 +81,17 @@ class MeasurementGetQueryConfig(BaseModel):
model_config = ConfigDict(extra="forbid", populate_by_name=True)
- columns: Optional[str] = Field(default=None,
pattern=_regex_comma_separated_string)
- end_date: Optional[StrictInt] = Field(default=None, alias="endDate")
- limit: Optional[int] = Field(ge=1, default=1000)
- offset: Optional[int] = Field(default=None, ge=0)
- order: Optional[Literal["ASC", "DESC"]] = None
- page_no: Optional[int] = Field(default=None, alias="page", ge=1)
- start_date: Optional[StrictInt] = Field(default=None, alias="startDate")
+ columns: str | None = Field(default=None,
pattern=_regex_comma_separated_string)
+ end_date: StrictInt | None = Field(default=None, alias="endDate")
+ limit: int | None = Field(ge=1, default=1000)
+ offset: int | None = Field(default=None, ge=0)
+ order: Literal["ASC", "DESC"] | None = None
+ page_no: int | None = Field(default=None, alias="page", ge=1)
+ start_date: StrictInt | None = Field(default=None, alias="startDate")
@field_validator("columns", mode="before")
@classmethod
- def _convert_to_comma_separated_string(cls, value: Any) -> Optional[str]:
+ def _convert_to_comma_separated_string(cls, value: Any) -> str | None:
"""Pydantic validator to convert a list to a comma separated string.
This is necessary for the StreamPipes API.
@@ -283,7 +283,7 @@ class DataLakeMeasureEndpoint(APIEndpoint):
"""
@staticmethod
- def _validate_query_params(query_params: Dict[str, Any]) ->
MeasurementGetQueryConfig:
+ def _validate_query_params(query_params: dict[str, Any]) ->
MeasurementGetQueryConfig:
"""Validates given query params.
Validates the given query parameters via the
@@ -314,7 +314,7 @@ class DataLakeMeasureEndpoint(APIEndpoint):
return config
@property
- def _resource_cls(self) -> Type[QueryResult]:
+ def _resource_cls(self) -> type[QueryResult]:
"""
Additional reference to resource class.
This endpoint deviates from the desired relationship
@@ -329,7 +329,7 @@ class DataLakeMeasureEndpoint(APIEndpoint):
return QueryResult
@property
- def _container_cls(self) -> Type[ResourceContainer]:
+ def _container_cls(self) -> type[ResourceContainer]:
"""Defines the model container class the endpoint refers to.
Returns
@@ -339,14 +339,14 @@ class DataLakeMeasureEndpoint(APIEndpoint):
return DataLakeMeasures
@property
- def _relative_api_path(self) -> Tuple[str, ...]:
+ def _relative_api_path(self) -> tuple[str, ...]:
"""Defines the relative api path to the DataLakeMeasurement endpoint.
Each path within the URL is defined as an own string.
"""
return "api", "v4", "datalake", "measurements"
- def get(self, identifier: str, **kwargs: Optional[Dict[str, Any]]) ->
QueryResult:
+ def get(self, identifier: str, **kwargs: dict[str, Any] | None) ->
QueryResult:
"""Queries the specified data lake measure from the API.
By default, the maximum number of returned records is 1000.
diff --git a/streampipes-client-python/streampipes/endpoint/api/data_stream.py
b/streampipes-client-python/streampipes/endpoint/api/data_stream.py
index 6067e0b35d..3987db626f 100644
--- a/streampipes-client-python/streampipes/endpoint/api/data_stream.py
+++ b/streampipes-client-python/streampipes/endpoint/api/data_stream.py
@@ -22,7 +22,6 @@ Specific implementation of the StreamPipes API's data stream
endpoints.
__all__ = [
"DataStreamEndpoint",
]
-from typing import Tuple, Type
from streampipes.endpoint.endpoint import APIEndpoint
from streampipes.model.container import DataStreams
@@ -64,7 +63,7 @@ class DataStreamEndpoint(APIEndpoint):
"""
@property
- def _container_cls(self) -> Type[ResourceContainer]:
+ def _container_cls(self) -> type[ResourceContainer]:
"""Defines the model container class the endpoint refers to.
@@ -75,7 +74,7 @@ class DataStreamEndpoint(APIEndpoint):
return DataStreams
@property
- def _relative_api_path(self) -> Tuple[str, ...]:
+ def _relative_api_path(self) -> tuple[str, ...]:
"""Defines the relative api path to the DataStream endpoint.
Each path within the URL is defined as an own string.
diff --git a/streampipes-client-python/streampipes/endpoint/api/version.py
b/streampipes-client-python/streampipes/endpoint/api/version.py
index 37fc4fc6be..0fc68f95dd 100644
--- a/streampipes-client-python/streampipes/endpoint/api/version.py
+++ b/streampipes-client-python/streampipes/endpoint/api/version.py
@@ -23,7 +23,6 @@ __all__ = [
"VersionEndpoint",
]
-from typing import Tuple, Type
from streampipes.endpoint import APIEndpoint
from streampipes.model.container import Versions
@@ -64,7 +63,7 @@ class VersionEndpoint(APIEndpoint):
"""
@property
- def _container_cls(self) -> Type[ResourceContainer]:
+ def _container_cls(self) -> type[ResourceContainer]:
"""Defines the model container class the endpoint refers to.
Returns
@@ -75,7 +74,7 @@ class VersionEndpoint(APIEndpoint):
return Versions
@property
- def _resource_cls(cls) -> Type[Version]:
+ def _resource_cls(cls) -> type[Version]:
"""Returns the class of the resource that are bundled.
Returns
@@ -85,7 +84,7 @@ class VersionEndpoint(APIEndpoint):
return Version
@property
- def _relative_api_path(self) -> Tuple[str, ...]:
+ def _relative_api_path(self) -> tuple[str, ...]:
"""Defines the relative api path to the DataStream endpoint.
Each path within the URL is defined as an own string.
diff --git a/streampipes-client-python/streampipes/endpoint/endpoint.py
b/streampipes-client-python/streampipes/endpoint/endpoint.py
index d5f44d118a..a29a04a63c 100644
--- a/streampipes-client-python/streampipes/endpoint/endpoint.py
+++ b/streampipes-client-python/streampipes/endpoint/endpoint.py
@@ -24,8 +24,9 @@ An endpoint provides all options to communicate with a
dedicated part of StreamP
import json
import logging
from abc import ABC, abstractmethod
+from collections.abc import Callable
from http import HTTPStatus
-from typing import Callable, Optional, Tuple, Type, final
+from typing import final
from requests import Response
from requests.exceptions import HTTPError
@@ -87,7 +88,7 @@ class APIEndpoint(Endpoint):
@property
@abstractmethod
- def _container_cls(self) -> Type[ResourceContainer]:
+ def _container_cls(self) -> type[ResourceContainer]:
"""Defines the model container class the endpoint refers to.
This model container class corresponds to the Python data model,
which handles multiple resources returned from the endpoint.
@@ -101,7 +102,7 @@ class APIEndpoint(Endpoint):
@property
@abstractmethod
- def _relative_api_path(self) -> Tuple[str, ...]:
+ def _relative_api_path(self) -> tuple[str, ...]:
"""Defines the relative api path with regard to the StreamPipes API
URL.
Each path within the URL is defined as an own string.
@@ -237,7 +238,7 @@ class APIEndpoint(Endpoint):
headers={"Content-type": "application/json"},
)
- def put(self, resource: Resource, identifier: Optional[str] = None) ->
None:
+ def put(self, resource: Resource, identifier: str | None = None) -> None:
"""Allows to update a resource in the StreamPipes API.
Parameters
@@ -276,7 +277,7 @@ class MessagingEndpoint(Endpoint):
"""
def __init__(self, parent_client: "StreamPipesClient"): # type: ignore #
noqa: F821
- self._broker: Optional[Broker] = None
+ self._broker: Broker | None = None
super().__init__(parent_client=parent_client)
@property
diff --git
a/streampipes-client-python/streampipes/function_zoo/river_function.py
b/streampipes-client-python/streampipes/function_zoo/river_function.py
index 01c475bd35..456ed49ecb 100644
--- a/streampipes-client-python/streampipes/function_zoo/river_function.py
+++ b/streampipes-client-python/streampipes/function_zoo/river_function.py
@@ -14,7 +14,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from typing import Any, Callable, Dict, List, Optional
+from collections.abc import Callable
+from typing import Any
from streampipes.client.client import StreamPipesClient
from streampipes.functions.broker.broker_handler import get_broker_description
@@ -59,9 +60,9 @@ class RiverFunction(StreamPipesFunction):
function_definition: FunctionDefinition,
model: Any,
supervised: bool,
- target_label: Optional[str],
+ target_label: str | None,
on_start: Callable[[Any, FunctionContext], None],
- on_event: Callable[[Any, Dict[str, Any], str], None],
+ on_event: Callable[[Any, dict[str, Any], str], None],
on_stop: Callable[[Any], None],
) -> None:
super().__init__(function_definition)
@@ -89,7 +90,7 @@ class RiverFunction(StreamPipesFunction):
"""
self.on_start(self, context)
- def onEvent(self, event: Dict[str, Any], streamId: str):
+ def onEvent(self, event: dict[str, Any], streamId: str):
"""Trains the model with the incoming events and sends the prediction
back to StreamPipes.
Parameters
@@ -159,14 +160,14 @@ class OnlineML:
def __init__(
self,
client: StreamPipesClient,
- stream_ids: List[str],
+ stream_ids: list[str],
model: Any,
output_stream_name: str = "Online ML Prediction",
prediction_type: str = RuntimeType.STRING.value,
supervised: bool = False,
- target_label: Optional[str] = None,
+ target_label: str | None = None,
on_start: Callable[[Any, FunctionContext], None] = lambda self,
context: None,
- on_event: Callable[[Any, Dict[str, Any], str], None] = lambda self,
event, streamId: None,
+ on_event: Callable[[Any, dict[str, Any], str], None] = lambda self,
event, streamId: None,
on_stop: Callable[[Any], None] = lambda self: None,
):
self.client = client
diff --git a/streampipes-client-python/streampipes/functions/broker/consumer.py
b/streampipes-client-python/streampipes/functions/broker/consumer.py
index 03e9885042..43549ba518 100644
--- a/streampipes-client-python/streampipes/functions/broker/consumer.py
+++ b/streampipes-client-python/streampipes/functions/broker/consumer.py
@@ -15,7 +15,7 @@
# limitations under the License.
#
from abc import abstractmethod
-from typing import AsyncIterator
+from collections.abc import AsyncIterator
from streampipes.functions.broker import Broker
from streampipes.model.resource.data_stream import DataStream
diff --git
a/streampipes-client-python/streampipes/functions/broker/kafka/kafka_consumer.py
b/streampipes-client-python/streampipes/functions/broker/kafka/kafka_consumer.py
index ff29b033a0..35ac6e0821 100644
---
a/streampipes-client-python/streampipes/functions/broker/kafka/kafka_consumer.py
+++
b/streampipes-client-python/streampipes/functions/broker/kafka/kafka_consumer.py
@@ -16,7 +16,7 @@
#
import logging
-from typing import AsyncIterator
+from collections.abc import AsyncIterator
from confluent_kafka import Consumer as KafkaConnection # type: ignore
diff --git
a/streampipes-client-python/streampipes/functions/broker/kafka/kafka_publisher.py
b/streampipes-client-python/streampipes/functions/broker/kafka/kafka_publisher.py
index 94635f16a0..cb4141d783 100644
---
a/streampipes-client-python/streampipes/functions/broker/kafka/kafka_publisher.py
+++
b/streampipes-client-python/streampipes/functions/broker/kafka/kafka_publisher.py
@@ -17,7 +17,7 @@
import json
import logging
-from typing import Any, Dict
+from typing import Any
from confluent_kafka import Producer # type: ignore
@@ -48,7 +48,7 @@ class KafkaPublisher(Publisher):
self.kafka_producer = Producer({"bootstrap.servers":
f"{hostname}:{port}"})
logger.info(f"Connecting to Kafka at {hostname}:{port}")
- async def publish_event(self, event: Dict[str, Any]):
+ async def publish_event(self, event: dict[str, Any]):
"""Publish an event to a connected data stream.
Parameters
diff --git
a/streampipes-client-python/streampipes/functions/broker/nats/nats_consumer.py
b/streampipes-client-python/streampipes/functions/broker/nats/nats_consumer.py
index 895a58147a..44dc4da44d 100644
---
a/streampipes-client-python/streampipes/functions/broker/nats/nats_consumer.py
+++
b/streampipes-client-python/streampipes/functions/broker/nats/nats_consumer.py
@@ -15,7 +15,7 @@
# limitations under the License.
#
import logging
-from typing import AsyncIterator
+from collections.abc import AsyncIterator
from nats import connect
diff --git
a/streampipes-client-python/streampipes/functions/broker/nats/nats_publisher.py
b/streampipes-client-python/streampipes/functions/broker/nats/nats_publisher.py
index 5c38c3787a..a5ffd99012 100644
---
a/streampipes-client-python/streampipes/functions/broker/nats/nats_publisher.py
+++
b/streampipes-client-python/streampipes/functions/broker/nats/nats_publisher.py
@@ -16,7 +16,7 @@
#
import json
import logging
-from typing import Any, Dict
+from typing import Any
from nats import connect
@@ -48,7 +48,7 @@ class NatsPublisher(Publisher):
self.nats_client = await connect([f"nats://{hostname}:{port}"])
logger.info(f"Connecting to NATS at {hostname}:{port}")
- async def publish_event(self, event: Dict[str, Any]):
+ async def publish_event(self, event: dict[str, Any]):
"""Publish an event to a connected data stream.
Parameters
diff --git
a/streampipes-client-python/streampipes/functions/broker/output_collector.py
b/streampipes-client-python/streampipes/functions/broker/output_collector.py
index a819539aaa..909517bb9b 100644
--- a/streampipes-client-python/streampipes/functions/broker/output_collector.py
+++ b/streampipes-client-python/streampipes/functions/broker/output_collector.py
@@ -15,7 +15,8 @@
# limitations under the License.
#
import asyncio
-from typing import Any, Coroutine, Dict
+from collections.abc import Coroutine
+from typing import Any
from streampipes.functions.broker import Publisher, get_broker
from streampipes.model.resource.data_stream import DataStream
@@ -41,7 +42,7 @@ class OutputCollector:
self.publisher: Publisher = get_broker(data_stream, is_publisher=True)
# type: ignore
self._run_coroutine(self.publisher.connect(data_stream))
- def collect(self, event: Dict[str, Any]) -> None:
+ def collect(self, event: dict[str, Any]) -> None:
"""Publishes an event to the output stream.
Parameters
diff --git
a/streampipes-client-python/streampipes/functions/broker/publisher.py
b/streampipes-client-python/streampipes/functions/broker/publisher.py
index bdcb0d8e89..49129feba6 100644
--- a/streampipes-client-python/streampipes/functions/broker/publisher.py
+++ b/streampipes-client-python/streampipes/functions/broker/publisher.py
@@ -15,7 +15,7 @@
# limitations under the License.
#
from abc import abstractmethod
-from typing import Any, Dict
+from typing import Any
from streampipes.functions.broker import Broker
@@ -27,7 +27,7 @@ class Publisher(Broker):
"""
@abstractmethod
- async def publish_event(self, event: Dict[str, Any]) -> None:
+ async def publish_event(self, event: dict[str, Any]) -> None:
"""Publish an event to a connected data stream.
Parameters
diff --git
a/streampipes-client-python/streampipes/functions/function_handler.py
b/streampipes-client-python/streampipes/functions/function_handler.py
index 12361a3d13..49c04b70f9 100644
--- a/streampipes-client-python/streampipes/functions/function_handler.py
+++ b/streampipes-client-python/streampipes/functions/function_handler.py
@@ -17,8 +17,8 @@
import asyncio
import json
import logging
+from collections.abc import AsyncIterator
from http import HTTPStatus
-from typing import AsyncIterator, Dict, List
from requests import HTTPError
@@ -57,8 +57,8 @@ class FunctionHandler:
def __init__(self, registration: Registration, client: StreamPipesClient)
-> None:
self.registration = registration
self.client = client
- self.stream_contexts: Dict[str, DataStreamContext] = {}
- self.brokers: List[Broker] = []
+ self.stream_contexts: dict[str, DataStreamContext] = {}
+ self.brokers: list[Broker] = []
def initializeFunctions(self) -> None:
"""Creates the context for every data stream and starts the event loop
to manage the StreamPipes Functions.
@@ -116,8 +116,8 @@ class FunctionHandler:
-------
None
"""
- messages: Dict[str, AsyncIterator] = dict()
- contexts: Dict[str, FunctionContext] = dict()
+ messages: dict[str, AsyncIterator] = dict()
+ contexts: dict[str, FunctionContext] = dict()
for stream_id in self.stream_contexts.keys():
data_stream = self.stream_contexts[stream_id].schema
diff --git a/streampipes-client-python/streampipes/functions/registration.py
b/streampipes-client-python/streampipes/functions/registration.py
index 5cbd39d0c9..979410c912 100644
--- a/streampipes-client-python/streampipes/functions/registration.py
+++ b/streampipes-client-python/streampipes/functions/registration.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from typing import List
from streampipes.functions.streampipes_function import StreamPipesFunction
@@ -30,7 +29,7 @@ class Registration:
"""
def __init__(self) -> None:
- self.functions: List[StreamPipesFunction] = []
+ self.functions: list[StreamPipesFunction] = []
def register(self, streampipes_function: StreamPipesFunction):
"""Registers a new function.
@@ -48,7 +47,7 @@ class Registration:
self.functions.append(streampipes_function) # TODO register function
to AdminAPI
return self
- def getFunctions(self) -> List[StreamPipesFunction]:
+ def getFunctions(self) -> list[StreamPipesFunction]:
"""Get all registered functions.
This method exists to be consistent with the Java client.
diff --git
a/streampipes-client-python/streampipes/functions/streampipes_function.py
b/streampipes-client-python/streampipes/functions/streampipes_function.py
index 8d3c063768..cec3316a3d 100644
--- a/streampipes-client-python/streampipes/functions/streampipes_function.py
+++ b/streampipes-client-python/streampipes/functions/streampipes_function.py
@@ -16,7 +16,7 @@
#
from abc import ABC, abstractmethod
from time import time
-from typing import Any, Dict, List, Optional
+from typing import Any
from streampipes.functions.broker.output_collector import OutputCollector
from streampipes.functions.utils.function_context import FunctionContext
@@ -42,14 +42,14 @@ class StreamPipesFunction(ABC):
List of all output collectors which are created based on the provided
function definitions.
"""
- def __init__(self, function_definition: Optional[FunctionDefinition] =
None):
+ def __init__(self, function_definition: FunctionDefinition | None = None):
self.function_definition = function_definition or FunctionDefinition()
self.output_collectors = {
stream_id: OutputCollector(data_stream)
for stream_id, data_stream in
self.function_definition.output_data_streams.items()
}
- def add_output(self, stream_id: str, event: Dict[str, Any]):
+ def add_output(self, stream_id: str, event: dict[str, Any]):
"""Send an event via an output data stream to StreamPipes.
Parameters
@@ -84,7 +84,7 @@ class StreamPipesFunction(ABC):
collector.disconnect()
self.onServiceStopped()
- def requiredStreamIds(self) -> List[str]:
+ def requiredStreamIds(self) -> list[str]:
"""Get the ids of the streams needed by the function.
Returns
@@ -110,7 +110,7 @@ class StreamPipesFunction(ABC):
raise NotImplementedError # pragma: no cover
@abstractmethod
- def onEvent(self, event: Dict[str, Any], streamId: str) -> None:
+ def onEvent(self, event: dict[str, Any], streamId: str) -> None:
"""Is called for every event of a data stream.
Parameters
diff --git
a/streampipes-client-python/streampipes/functions/utils/async_iter_handler.py
b/streampipes-client-python/streampipes/functions/utils/async_iter_handler.py
index 0c90e7859f..6087a69847 100644
---
a/streampipes-client-python/streampipes/functions/utils/async_iter_handler.py
+++
b/streampipes-client-python/streampipes/functions/utils/async_iter_handler.py
@@ -15,14 +15,15 @@
# limitations under the License.
#
import asyncio
-from typing import Any, AsyncGenerator, AsyncIterator, Dict, Tuple
+from collections.abc import AsyncGenerator, AsyncIterator
+from typing import Any
class AsyncIterHandler:
"""Handles asynchronous iterators to get every message after another in
parallel."""
@staticmethod
- async def anext(stream_id: str, message: AsyncIterator) -> Tuple[str, Any]:
+ async def anext(stream_id: str, message: AsyncIterator) -> tuple[str, Any]:
"""Gets the next message from an AsyncIterator.
Parameters
@@ -43,7 +44,7 @@ class AsyncIterHandler:
return "stop", None
@staticmethod
- async def combine_async_messages(messages: Dict[str, AsyncIterator]) ->
AsyncGenerator:
+ async def combine_async_messages(messages: dict[str, AsyncIterator]) ->
AsyncGenerator:
"""Continuously gets the next published message from multiple
AsyncIterators in parallel.
Parameters
diff --git
a/streampipes-client-python/streampipes/functions/utils/data_stream_context.py
b/streampipes-client-python/streampipes/functions/utils/data_stream_context.py
index 315a13cd0a..026e1001b1 100644
---
a/streampipes-client-python/streampipes/functions/utils/data_stream_context.py
+++
b/streampipes-client-python/streampipes/functions/utils/data_stream_context.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from typing import List
from streampipes.functions.broker import Consumer
from streampipes.functions.streampipes_function import StreamPipesFunction
@@ -34,7 +33,7 @@ class DataStreamContext:
The consumer to connect to this data stream.
"""
- def __init__(self, functions: List[StreamPipesFunction], schema:
DataStream, broker: Consumer) -> None:
+ def __init__(self, functions: list[StreamPipesFunction], schema:
DataStream, broker: Consumer) -> None:
self.functions = functions
self.schema = schema
self.broker = broker
diff --git
a/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
b/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
index 00dd200b03..c61b418d9d 100644
---
a/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
+++
b/streampipes-client-python/streampipes/functions/utils/data_stream_generator.py
@@ -16,7 +16,6 @@
#
from enum import Enum
-from typing import Dict, Optional
from streampipes.functions.broker import SupportedBroker
from streampipes.model.common import (
@@ -52,8 +51,8 @@ class RuntimeType(Enum):
# TODO Use an more general approach to create a data stream
def create_data_stream(
name: str,
- attributes: Dict[str, str],
- stream_id: Optional[str] = None,
+ attributes: dict[str, str],
+ stream_id: str | None = None,
broker: SupportedBroker = SupportedBroker.NATS,
):
"""Creates a data stream
diff --git
a/streampipes-client-python/streampipes/functions/utils/function_context.py
b/streampipes-client-python/streampipes/functions/utils/function_context.py
index a4ef371418..a0cf548af6 100644
--- a/streampipes-client-python/streampipes/functions/utils/function_context.py
+++ b/streampipes-client-python/streampipes/functions/utils/function_context.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from typing import Dict, List
from streampipes.client.client import StreamPipesClient
from streampipes.model.resource.data_stream import DataStream
@@ -35,7 +34,7 @@ class FunctionContext:
The ids of the streams needed by this function.
"""
- def __init__(self, function_id: str, schema: Dict[str, DataStream],
client: StreamPipesClient, streams: List[str]):
+ def __init__(self, function_id: str, schema: dict[str, DataStream],
client: StreamPipesClient, streams: list[str]):
self.function_id = function_id
self.schema = schema
self.client = client
diff --git a/streampipes-client-python/streampipes/model/common.py
b/streampipes-client-python/streampipes/model/common.py
index 5acd54ae6d..8cabcdaaad 100644
--- a/streampipes-client-python/streampipes/model/common.py
+++ b/streampipes-client-python/streampipes/model/common.py
@@ -21,7 +21,6 @@ Classes of the StreamPipes data model that are commonly
shared.
import random
import string
-from typing import List, Optional
from uuid import uuid4
from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr
@@ -78,7 +77,7 @@ class BasicModel(BaseModel):
class BaseElement(BasicModel):
"""Structure of a basic element in the StreamPipes Backend."""
- element_id: Optional[StrictStr] = None
+ element_id: StrictStr | None = None
class ValueSpecification(BasicModel):
@@ -86,11 +85,11 @@ class ValueSpecification(BasicModel):
Data model of an `ValueSpecification` in compliance with the StreamPipes
Backend.
"""
- class_name: Optional[StrictStr] = Field(default=None, alias="@class")
- element_id: Optional[StrictStr] = None
- min_value: Optional[int] = None
- max_value: Optional[int] = None
- step: Optional[float] = None
+ class_name: StrictStr | None = Field(default=None, alias="@class")
+ element_id: StrictStr | None = None
+ min_value: int | None = None
+ max_value: int | None = None
+ step: float | None = None
class EventProperty(BasicModel):
@@ -100,15 +99,15 @@ class EventProperty(BasicModel):
class_name: StrictStr = Field(alias="@class",
default="org.apache.streampipes.model.schema.EventPropertyPrimitive")
element_id: StrictStr = Field(default_factory=lambda:
f"sp:eventproperty:{random_letters(6)}")
- label: Optional[StrictStr] = None
- description: Optional[StrictStr] = None
+ label: StrictStr | None = None
+ description: StrictStr | None = None
runtime_name: StrictStr
- semantic_type: Optional[StrictStr] = None
- property_scope: Optional[StrictStr] = Field(default="MEASUREMENT_PROPERTY")
- runtime_id: Optional[StrictStr] = None
+ semantic_type: StrictStr | None = None
+ property_scope: StrictStr | None = Field(default="MEASUREMENT_PROPERTY")
+ runtime_id: StrictStr | None = None
runtime_type: StrictStr =
Field(default="http://www.w3.org/2001/XMLSchema#string";)
- measurement_unit: Optional[StrictStr] = None
- value_specification: Optional[ValueSpecification] = None
+ measurement_unit: StrictStr | None = None
+ value_specification: ValueSpecification | None = None
class EventSchema(BasicModel):
@@ -116,7 +115,7 @@ class EventSchema(BasicModel):
Data model of an `EventSchema` in compliance with the StreamPipes Backend.
"""
- event_properties: List[EventProperty]
+ event_properties: list[EventProperty]
class ApplicationLink(BasicModel):
@@ -124,13 +123,13 @@ class ApplicationLink(BasicModel):
Data model of an `ApplicationLink` in compliance with the StreamPipes
Backend.
"""
- class_name: Optional[StrictStr] = Field(default=None, alias="@class")
- element_id: Optional[StrictStr] = None
- application_name: Optional[StrictStr] = None
- application_description: Optional[StrictStr] = None
- application_url: Optional[StrictStr] = None
- application_icon_url: Optional[StrictStr] = None
- application_link_type: Optional[StrictStr] = None
+ class_name: StrictStr | None = Field(default=None, alias="@class")
+ element_id: StrictStr | None = None
+ application_name: StrictStr | None = None
+ application_description: StrictStr | None = None
+ application_url: StrictStr | None = None
+ application_icon_url: StrictStr | None = None
+ application_link_type: StrictStr | None = None
class TopicDefinition(BasicModel):
@@ -138,7 +137,7 @@ class TopicDefinition(BasicModel):
Data model of a `TopicDefinition` in compliance with the StreamPipes
Backend.
"""
- class_name: Optional[StrictStr] = Field(
+ class_name: StrictStr | None = Field(
alias="@class",
default="org.apache.streampipes.model.grounding.SimpleTopicDefinition"
)
actual_topic_name: StrictStr = Field(default_factory=lambda:
f"org.apache.streampipes.connect.{uuid4()}")
@@ -163,7 +162,7 @@ class TransportFormat(BasicModel):
Data model of a `TransportFormat` in compliance with the StreamPipes
Backend.
"""
- rdf_type: List[StrictStr] =
Field(default=["http://sepa.event-processing.org/sepa#json";])
+ rdf_type: list[StrictStr] =
Field(default=["http://sepa.event-processing.org/sepa#json";])
class EventGrounding(BasicModel):
@@ -171,8 +170,8 @@ class EventGrounding(BasicModel):
Data model of an `EventGrounding` in compliance to with StreamPipes
Backend.
"""
- transport_protocols: List[TransportProtocol] =
Field(default_factory=lambda: [TransportProtocol()])
- transport_formats: List[TransportFormat] = Field(default_factory=lambda:
[TransportFormat()])
+ transport_protocols: list[TransportProtocol] =
Field(default_factory=lambda: [TransportProtocol()])
+ transport_formats: list[TransportFormat] = Field(default_factory=lambda:
[TransportFormat()])
class MeasurementCapability(BasicModel):
@@ -180,8 +179,8 @@ class MeasurementCapability(BasicModel):
Data model of a `MeasurementCapability` in compliance with the StreamPipes
Backend.
"""
- capability: Optional[StrictStr] = None
- element_id: Optional[StrictStr] = None
+ capability: StrictStr | None = None
+ element_id: StrictStr | None = None
class MeasurementObject(BasicModel):
@@ -189,5 +188,5 @@ class MeasurementObject(BasicModel):
Data model of a `MeasurementObject` in compliance with the StreamPipes
Backend.
"""
- element_id: Optional[StrictStr] = None
- measures_object: Optional[StrictStr] = None
+ element_id: StrictStr | None = None
+ measures_object: StrictStr | None = None
diff --git
a/streampipes-client-python/streampipes/model/container/data_lake_measures.py
b/streampipes-client-python/streampipes/model/container/data_lake_measures.py
index 8cb5bd159d..ac85f4d2c0 100644
---
a/streampipes-client-python/streampipes/model/container/data_lake_measures.py
+++
b/streampipes-client-python/streampipes/model/container/data_lake_measures.py
@@ -18,7 +18,6 @@
"""
Implementation of a resource container for the data lake measures endpoint.
"""
-from typing import Type
from streampipes.model.container.resource_container import ResourceContainer
from streampipes.model.resource.data_lake_measure import DataLakeMeasure
@@ -39,7 +38,7 @@ class DataLakeMeasures(ResourceContainer):
"""
@classmethod
- def _resource_cls(cls) -> Type[Resource]:
+ def _resource_cls(cls) -> type[Resource]:
"""Returns the class of the resource that are bundled.
Returns
diff --git
a/streampipes-client-python/streampipes/model/container/data_streams.py
b/streampipes-client-python/streampipes/model/container/data_streams.py
index 3112b75997..27594db345 100644
--- a/streampipes-client-python/streampipes/model/container/data_streams.py
+++ b/streampipes-client-python/streampipes/model/container/data_streams.py
@@ -18,7 +18,6 @@
"""
Implementation of a resource container for the data streams endpoint.
"""
-from typing import Type
from streampipes.model.container.resource_container import ResourceContainer
from streampipes.model.resource.data_stream import DataStream
@@ -39,7 +38,7 @@ class DataStreams(ResourceContainer):
"""
@classmethod
- def _resource_cls(cls) -> Type[Resource]:
+ def _resource_cls(cls) -> type[Resource]:
"""Returns the class of the resource that are bundled.
Returns
diff --git
a/streampipes-client-python/streampipes/model/container/resource_container.py
b/streampipes-client-python/streampipes/model/container/resource_container.py
index 9897f396a9..355a946066 100644
---
a/streampipes-client-python/streampipes/model/container/resource_container.py
+++
b/streampipes-client-python/streampipes/model/container/resource_container.py
@@ -27,7 +27,6 @@ from __future__ import annotations
import json
from abc import ABC, abstractmethod
-from typing import Dict, List, Type
import pandas as pd
from pydantic import ValidationError
@@ -131,7 +130,7 @@ class ResourceContainer(ABC):
"""
- def __init__(self, resources: List[Resource]):
+ def __init__(self, resources: list[Resource]):
self._resources = resources
def __getitem__(self, position: int) -> Resource:
@@ -146,7 +145,7 @@ class ResourceContainer(ABC):
@classmethod
@abstractmethod
- def _resource_cls(cls) -> Type[Resource]:
+ def _resource_cls(cls) -> type[Resource]:
"""Returns the class of the resource that are bundled.
Returns
@@ -193,7 +192,7 @@ class ResourceContainer(ABC):
return resource_container
- def to_dicts(self, use_source_names: bool = False) -> List[Dict]:
+ def to_dicts(self, use_source_names: bool = False) -> list[dict]:
"""Returns the contained resources as list of dictionaries.
Parameters
diff --git a/streampipes-client-python/streampipes/model/container/versions.py
b/streampipes-client-python/streampipes/model/container/versions.py
index f2e952fb34..2326a46500 100644
--- a/streampipes-client-python/streampipes/model/container/versions.py
+++ b/streampipes-client-python/streampipes/model/container/versions.py
@@ -23,7 +23,6 @@ __all__ = [
"Versions",
]
-from typing import Type
from streampipes.model.container.resource_container import ResourceContainer
from streampipes.model.resource import Version
@@ -45,7 +44,7 @@ class Versions(ResourceContainer):
"""
@classmethod
- def _resource_cls(cls) -> Type[Resource]:
+ def _resource_cls(cls) -> type[Resource]:
"""Returns the class of the resource that are bundled.
Returns
diff --git
a/streampipes-client-python/streampipes/model/resource/data_lake_measure.py
b/streampipes-client-python/streampipes/model/resource/data_lake_measure.py
index e2af36b5b1..ccbe47c6fb 100644
--- a/streampipes-client-python/streampipes/model/resource/data_lake_measure.py
+++ b/streampipes-client-python/streampipes/model/resource/data_lake_measure.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from typing import Optional
from pydantic import StrictBool, StrictStr
@@ -56,11 +55,11 @@ class DataLakeMeasure(Resource):
"num_event_properties": len(self.event_schema.event_properties) if
self.event_schema else 0,
}
- element_id: Optional[StrictStr] = None
+ element_id: StrictStr | None = None
measure_name: StrictStr
timestamp_field: StrictStr
- event_schema: Optional[EventSchema] = None
- pipeline_id: Optional[StrictStr] = None
- pipeline_name: Optional[StrictStr] = None
+ event_schema: EventSchema | None = None
+ pipeline_id: StrictStr | None = None
+ pipeline_name: StrictStr | None = None
pipeline_is_running: StrictBool
- schema_version: Optional[StrictStr] = None
+ schema_version: StrictStr | None = None
diff --git
a/streampipes-client-python/streampipes/model/resource/data_series.py
b/streampipes-client-python/streampipes/model/resource/data_series.py
index 8bec7310f5..651457dab2 100644
--- a/streampipes-client-python/streampipes/model/resource/data_series.py
+++ b/streampipes-client-python/streampipes/model/resource/data_series.py
@@ -18,7 +18,7 @@
from __future__ import annotations
import json
-from typing import Any, Dict, List, Optional, Union
+from typing import Any
import pandas as pd
from pydantic import StrictInt, StrictStr
@@ -84,7 +84,7 @@ class DataSeries(Resource):
return cls.model_validate(data_series)
- def convert_to_pandas_representation(self) -> Dict[str, Union[List[str],
List[List[Any]]]]:
+ def convert_to_pandas_representation(self) -> dict[str, list[str] |
list[list[Any]]]:
"""Returns the dictionary representation of a data lake series
to be used when creating a pandas Dataframe.
@@ -99,9 +99,9 @@ class DataSeries(Resource):
return self.model_dump(include={"headers", "rows"})
total: StrictInt
- headers: List[StrictStr]
- rows: List[List[Any]]
- tags: Optional[str] = None
+ headers: list[StrictStr]
+ rows: list[list[Any]]
+ tags: str | None = None
def to_pandas(self) -> pd.DataFrame:
"""Returns the data lake series in representation of a Pandas
Dataframe.
diff --git
a/streampipes-client-python/streampipes/model/resource/data_stream.py
b/streampipes-client-python/streampipes/model/resource/data_stream.py
index a75df7a79e..193c0a6a33 100644
--- a/streampipes-client-python/streampipes/model/resource/data_stream.py
+++ b/streampipes-client-python/streampipes/model/resource/data_stream.py
@@ -14,7 +14,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
-from typing import List, Optional
from pydantic import Field, StrictBool, StrictStr
@@ -88,25 +87,25 @@ class DataStream(Resource):
class_name: StrictStr = Field(alias="@class", default_factory=lambda:
"org.apache.streampipes.model.SpDataStream")
element_id: StrictStr = Field(default="")
name: StrictStr = Field(default="Unnamed")
- description: Optional[StrictStr] = None
- icon_url: Optional[StrictStr] = None
- app_id: Optional[StrictStr] = None
+ description: StrictStr | None = None
+ icon_url: StrictStr | None = None
+ app_id: StrictStr | None = None
includes_assets: StrictBool = Field(default=False)
includes_locales: StrictBool = Field(default=False)
- included_assets: List[StrictStr] = Field(default_factory=list)
- included_locales: List[StrictStr] = Field(default_factory=list)
- application_links: List[ApplicationLink] = Field(default_factory=list)
+ included_assets: list[StrictStr] = Field(default_factory=list)
+ included_locales: list[StrictStr] = Field(default_factory=list)
+ application_links: list[ApplicationLink] = Field(default_factory=list)
internally_managed: StrictBool = Field(default=False)
- connected_to: Optional[List[StrictStr]] = None
+ connected_to: list[StrictStr] | None = None
event_grounding: EventGrounding = Field(default_factory=EventGrounding)
- event_schema: Optional[EventSchema] = None
- measurement_capability: Optional[List[MeasurementCapability]] = None
- measurement_object: Optional[List[MeasurementObject]] = None
- corresponding_adapter_id: Optional[StrictStr] = None
- category: Optional[List[StrictStr]] = None
- uri: Optional[StrictStr] = None
- dom: Optional[StrictStr] = None
- rev: Optional[StrictStr] = Field(default=None, alias="_rev")
+ event_schema: EventSchema | None = None
+ measurement_capability: list[MeasurementCapability] | None = None
+ measurement_object: list[MeasurementObject] | None = None
+ corresponding_adapter_id: StrictStr | None = None
+ category: list[StrictStr] | None = None
+ uri: StrictStr | None = None
+ dom: StrictStr | None = None
+ rev: StrictStr | None = Field(default=None, alias="_rev")
def to_dict(self, use_source_names=True):
"""Returns the resource in dictionary representation.
diff --git a/streampipes-client-python/streampipes/model/resource/exceptions.py
b/streampipes-client-python/streampipes/model/resource/exceptions.py
index 88bb9e007a..f42b6e3c48 100644
--- a/streampipes-client-python/streampipes/model/resource/exceptions.py
+++ b/streampipes-client-python/streampipes/model/resource/exceptions.py
@@ -15,7 +15,6 @@
# limitations under the License.
#
-from typing import Optional
class StreamPipesUnsupportedDataSeries(Exception):
@@ -23,7 +22,7 @@ class StreamPipesUnsupportedDataSeries(Exception):
cannot be parsed with the current implementation of the resource.
"""
- def __init__(self, reason: Optional[str] = None):
+ def __init__(self, reason: str | None = None):
super().__init__(
"The Data Lake series returned by the API appears "
"to have a structure that is not currently supported by the Python
client."
diff --git
a/streampipes-client-python/streampipes/model/resource/function_definition.py
b/streampipes-client-python/streampipes/model/resource/function_definition.py
index c4bdad8a99..79522450e6 100644
---
a/streampipes-client-python/streampipes/model/resource/function_definition.py
+++
b/streampipes-client-python/streampipes/model/resource/function_definition.py
@@ -19,7 +19,6 @@ __all__ = [
"FunctionDefinition",
]
-from typing import Dict, List
from uuid import uuid4
from pydantic import Field, StrictInt, StrictStr
@@ -68,10 +67,10 @@ class FunctionDefinition(Resource):
"""
function_id: FunctionId = Field(default_factory=FunctionId)
- consumed_streams: List[str] = Field(default_factory=list)
- output_data_streams: Dict[str, DataStream] = Field(default_factory=dict)
+ consumed_streams: list[str] = Field(default_factory=list)
+ output_data_streams: dict[str, DataStream] = Field(default_factory=dict)
- def convert_to_pandas_representation(self) -> Dict:
+ def convert_to_pandas_representation(self) -> dict:
"""Returns the dictionary representation of a function definition
to be used when creating a pandas Dataframe.
@@ -102,7 +101,7 @@ class FunctionDefinition(Resource):
self.output_data_streams[data_stream.element_id] = data_stream
return self
- def get_output_data_streams(self) -> Dict[str, DataStream]:
+ def get_output_data_streams(self) -> dict[str, DataStream]:
"""Get the output data streams of the function.
Returns
@@ -114,7 +113,7 @@ class FunctionDefinition(Resource):
return self.output_data_streams
- def get_output_stream_ids(self) -> List[str]:
+ def get_output_stream_ids(self) -> list[str]:
"""Get the stream ids of the output data streams.
Returns
diff --git
a/streampipes-client-python/streampipes/model/resource/query_result.py
b/streampipes-client-python/streampipes/model/resource/query_result.py
index 5159e3969d..fa714649f8 100644
--- a/streampipes-client-python/streampipes/model/resource/query_result.py
+++ b/streampipes-client-python/streampipes/model/resource/query_result.py
@@ -16,7 +16,7 @@
#
from itertools import chain
-from typing import Any, Dict, List, Literal, Optional, Union
+from typing import Any, Literal
import pandas as pd
from pydantic import Field, StrictInt, StrictStr
@@ -38,7 +38,7 @@ class QueryResult(Resource):
the Python representation (both serialized and deserialized) and Java
representation (serialized only).
"""
- def convert_to_pandas_representation(self) -> Dict[str, Union[List[str],
List[List[Any]]]]:
+ def convert_to_pandas_representation(self) -> dict[str, list[str] |
list[list[Any]]]:
"""Returns the dictionary representation of a data lake series
to be used when creating a pandas Dataframe.
@@ -70,11 +70,11 @@ class QueryResult(Resource):
}
total: StrictInt
- headers: List[StrictStr]
- all_data_series: List[DataSeries]
+ headers: list[StrictStr]
+ all_data_series: list[DataSeries]
query_status: Literal["OK", "TOO_MUCH_DATA"] = Field(alias="spQueryStatus")
source_index: StrictInt
- for_id: Optional[str] = None
+ for_id: str | None = None
last_timestamp: StrictInt
def to_pandas(self) -> pd.DataFrame:
@@ -97,7 +97,7 @@ class QueryResult(Resource):
cls,
df: pd.DataFrame,
source_index: int = 0,
- for_id: Optional[str] = None,
+ for_id: str | None = None,
query_status: Literal["OK", "TOO_MUCH_DATA"] = "OK",
) -> "QueryResult":
"""Create a QueryResult object from a pandas DataFrame.
diff --git a/streampipes-client-python/streampipes/model/resource/resource.py
b/streampipes-client-python/streampipes/model/resource/resource.py
index 9042e5b7ba..20544bfc96 100644
--- a/streampipes-client-python/streampipes/model/resource/resource.py
+++ b/streampipes-client-python/streampipes/model/resource/resource.py
@@ -21,7 +21,6 @@ General and abstract implementation for a resource.
A resource defines the data model that is used by a resource container
(`model.container.resourceContainer`).
"""
from abc import abstractmethod
-from typing import Dict
from streampipes.model.common import BasicModel
@@ -41,7 +40,7 @@ class Resource(BasicModel):
"""
@abstractmethod
- def convert_to_pandas_representation(self) -> Dict:
+ def convert_to_pandas_representation(self) -> dict:
"""Returns a dictionary representation to be used when creating a
pandas Dataframe.
Returns
diff --git a/streampipes-client-python/streampipes/model/resource/version.py
b/streampipes-client-python/streampipes/model/resource/version.py
index 1a03c5c7a4..04644cf6f6 100644
--- a/streampipes-client-python/streampipes/model/resource/version.py
+++ b/streampipes-client-python/streampipes/model/resource/version.py
@@ -19,7 +19,6 @@ __all__ = [
"Version",
]
-from typing import Dict, Optional
from pydantic import StrictStr, field_validator
@@ -35,13 +34,13 @@ class Version(Resource):
version of the StreamPipes backend the client is connected to
"""
- def convert_to_pandas_representation(self) -> Dict:
+ def convert_to_pandas_representation(self) -> dict:
"""Returns the dictionary representation of the version metadata
to be used when creating a pandas Dataframe.
"""
return self.to_dict(use_source_names=False)
- backend_version: Optional[StrictStr] = None
+ backend_version: StrictStr | None = None
@field_validator("backend_version", mode="before")
@classmethod
diff --git a/streampipes-client-python/tests/client/test_endpoint.py
b/streampipes-client-python/tests/client/test_endpoint.py
index 45af2db3bf..c5dc5fa8ff 100644
--- a/streampipes-client-python/tests/client/test_endpoint.py
+++ b/streampipes-client-python/tests/client/test_endpoint.py
@@ -17,7 +17,6 @@
import json
from copy import deepcopy
-from typing import Dict, List
from unittest import TestCase
from unittest.mock import MagicMock, call, patch
@@ -88,7 +87,7 @@ class TestStreamPipesEndpoints(TestCase):
}
]
- self.data_stream_all: List[Dict] = [
+ self.data_stream_all: list[dict] = [
{
"@class": "org.apache.streampipes.model.SpDataStream",
"elementId": "urn:streampipes.apache.org:eventstream:uPDKLI",
