optimized code (#3)

* optimized code

Author: mic1on

.github/workflows/test.yml

@@ -18,6 +18,7 @@ jobs:
         os: [ "ubuntu-latest" ]
         python-version: [ "3.8" ]
     runs-on: ${{ matrix.os }}
+    environment: test
     steps:
       - uses: actions/checkout@v2
       - name: Set up Python ${{ matrix.python-version }}
@@ -46,4 +47,6 @@ jobs:
           source .venv/bin/activate
           pytest tests/
         env:
-          SERVER_ADDR: ${{ secrets.SERVER_ADDR }}
+          NACOS_SERVER_ADDR: ${{ secrets.NACOS_SERVER_ADDR }}
+          NACOS_USERNAME: ${{ secrets.NACOS_USERNAME }}
+          NACOS_PASSWORD: ${{ secrets.NACOS_PASSWORD }}

src/use_nacos/client.py

@@ -1,7 +1,8 @@
 import logging
+import os
 from abc import abstractmethod
 from json import JSONDecodeError
-from typing import Union, Any, Optional, Dict, List, Generator
+from typing import Any, Optional, Dict, List, Generator
 
 import httpx
 from httpx import Request, Response, Auth, HTTPTransport, AsyncHTTPTransport
@@ -11,7 +12,7 @@
     ConfigAsyncEndpoint, InstanceAsyncEndpoint
 )
 from .exception import HTTPResponseError
-from .typings import SyncAsync
+from .typings import SyncAsync, HttpxClient
 
 logger = logging.getLogger(__name__)
 
@@ -23,18 +24,18 @@ class BaseClient:
 
     def __init__(
             self,
-            client: Union[httpx.Client, httpx.AsyncClient],
+            client: HttpxClient,
             server_addr: Optional[str] = None,
             username: Optional[str] = None,
             password: Optional[str] = None,
             namespace_id: Optional[str] = None,
     ):
-        self.server_addr = server_addr or DEFAULT_SERVER_ADDR
-        self.username = username
-        self.password = password
-        self.namespace_id = namespace_id or DEFAULT_NAMESPACE
+        self.server_addr = server_addr or os.environ.get("NACOS_SERVER_ADDR") or DEFAULT_SERVER_ADDR
+        self.username = username or os.environ.get("NACOS_USERNAME")
+        self.password = password or os.environ.get("NACOS_PASSWORD")
+        self.namespace_id = namespace_id or os.environ.get("NACOS_NAMESPACE") or DEFAULT_NAMESPACE
 
-        self._clients: List[Union[httpx.Client, httpx.AsyncClient]] = []
+        self._clients: List[HttpxClient] = []
         self.client = client
         # endpoints
         self.config = ConfigEndpoint(self)
@@ -43,11 +44,11 @@ def __init__(
         self.namespace = NamespaceEndpoint(self)
 
     @property
-    def client(self) -> Union[httpx.Client, httpx.AsyncClient]:
+    def client(self) -> HttpxClient:
         return self._clients[-1]
 
     @client.setter
-    def client(self, client: Union[httpx.Client, httpx.AsyncClient]) -> None:
+    def client(self, client: HttpxClient) -> None:
         client.base_url = httpx.URL(self.server_addr)
         client.timeout = httpx.Timeout(timeout=60_000 / 1_000)
         client.headers = httpx.Headers(
@@ -112,7 +113,11 @@ def __init__(
         """ Nacos Sync Client """
         client = client or httpx.Client(transport=HTTPTransport(retries=http_retries))
         super().__init__(
-            client, server_addr, username, password, namespace_id
+            client=client,
+            server_addr=server_addr,
+            username=username,
+            password=password,
+            namespace_id=namespace_id
         )
 
     def request(
@@ -152,7 +157,13 @@ def __init__(
     ):
         """ Nacos Async Client """
         client = client or httpx.AsyncClient(transport=AsyncHTTPTransport(retries=http_retries))
-        super().__init__(client, server_addr, username, password, namespace_id)
+        super().__init__(
+            client=client,
+            server_addr=server_addr,
+            username=username,
+            password=password,
+            namespace_id=namespace_id
+        )
         self.config = ConfigAsyncEndpoint(self)
         self.instance = InstanceAsyncEndpoint(self)
 

src/use_nacos/typings.py

@@ -1,7 +1,10 @@
 from typing import Awaitable, Union, TypeVar, TypedDict, Optional
 
+import httpx
+
 T = TypeVar("T")
 SyncAsync = Union[T, Awaitable[T]]
+HttpxClient = Union[httpx.Client, httpx.AsyncClient]
 
 
 class BeatType(TypedDict):

tests/test_config.py

@@ -1,5 +1,4 @@
 import json
-import os
 import random
 import time
 
@@ -12,17 +11,15 @@
 from use_nacos.exception import HTTPResponseError
 from use_nacos.serializer import JsonSerializer, AutoSerializer, YamlSerializer, TomlSerializer
 
-server_addr = os.environ.get('SERVER_ADDR')
-
 
 @pytest.fixture
 def client():
-    return NacosClient(server_addr=server_addr, username="nacos", password="nacos")
+    return NacosClient()
 
 
 @pytest.fixture
 def async_client():
-    return NacosAsyncClient(server_addr=server_addr, username="nacos", password="nacos")
+    return NacosAsyncClient()
 
 
 @pytest.fixture