Add Python 3.14 support

.github/workflows/test.yaml

@@ -12,7 +12,7 @@ jobs:
       runs-on: ubuntu-latest
       strategy:
         matrix:
-          python-version: ["3.10", "3.11", "3.12", "3.13"]
+          python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
           poetry-version: ["1.8.3"]
           pydantic-version: ["pydantic<2", "pydantic>=2"]
 

natsapi/applications.py

@@ -72,7 +72,7 @@ def __init__(
             self.loop = asyncio.get_running_loop()
             self._sharing_loop = True
         except RuntimeError:
-            self.loop = asyncio.get_event_loop()
+            self.loop = asyncio.new_event_loop()
             self._sharing_loop = False
 
         if app is not None:

natsapi/asyncapi/utils.py

@@ -1,3 +1,4 @@
+import types
 import typing
 from collections.abc import Sequence
 from enum import Enum
@@ -61,8 +62,9 @@ def get_flat_response_models(r) -> list[type[BaseModel]]:
 
     :r Single or multiple response models
     """
-    if type(r) is typing._UnionGenericAlias:
-        return list(r.__args__)
+    origin = typing.get_origin(r)
+    if origin in (typing.Union, types.UnionType):
+        return list(typing.get_args(r))
     else:
         return [r]
 

tests/asyncapi/test_generation.py

@@ -1,5 +1,4 @@
-import sys
-from typing import Any, Union
+from typing import Any, Optional, Union
 from uuid import uuid4
 
 import pytest
@@ -127,18 +126,18 @@ def test_generate_schema_w_external_docs_should_generate():
     assert schema["externalDocs"] == external_docs.dict()
 
 
-@pytest.mark.skipif(sys.version_info < (3, 10), reason="Requires Python 3.10+ for union syntax")
 async def test_optional_types_are_generated_correctly(app: NatsAPI):
     class User(BaseModel):
         name: str
         mandatory_property_1: str
         optional_property_1: str | None
-        optional_property_2: str | None
-        optional_property_3: str | None = None
-        optional_property_4: str | None | None = None
+        optional_property_2: Optional[str]  # noqa
+        optional_property_3: Optional[str] = None  # noqa
+        optional_property_4: Optional[str] | None = None  # noqa
         optional_property_5: str | None = None
         optional_property_6: str = None
         optional_property_7: str | int
+        optional_property_8: Union[str, None]
 
     user_router = SubjectRouter(prefix="v1", tags=["users"], deprecated=True)
 

tests/fixtures.py

@@ -30,4 +30,6 @@ async def app(client_config, event_loop):
 
 @pytest.fixture(scope="session")
 def event_loop():
-    yield asyncio.get_event_loop()
+    loop = asyncio.new_event_loop()
+    yield loop
+    loop.close()