tmp 5

Signed-off-by: kvmw <[email protected]>

Author: kvmw

greeter-messages/app.py

@@ -1,3 +1,5 @@
+"""Flask application for the greeter-messages service."""
+
 import os
 import logging
 from flask import Flask, request
@@ -11,15 +13,16 @@
 try:
     init_client()
     logger.info("Eureka client initialized successfully")
-except Exception as e:
-    logger.error(f"Failed to initialize Eureka client: {e}")
+except (ValueError, ImportError, OSError) as e:
+    logger.error("Failed to initialize Eureka client: %s", e)
 
 @app.route('/greeting')
 def greeting():
+    """Handle greeting requests."""
     salutation = request.args.get('salutation', 'Hello')
     name = request.args.get('name', 'Bob')
-    
-    logger.info(f"Greeting request: salutation='{salutation}', name='{name}'")
+
+    logger.info("Greeting request: salutation='%s', name='%s'", salutation, name)
     return f"{salutation}, {name}!"
 
 if __name__ == '__main__':

greeter-messages/eureka.py

@@ -1,12 +1,14 @@
+"""Eureka client integration for service discovery and registration."""
+
 import os
 import urllib.request
 import logging
+from typing import Union
+from urllib.error import URLError
+
 from cfenv import AppEnv
 from oauth2 import OAuth2Client
-import py_eureka_client.eureka_client as eureka_client
-import py_eureka_client.http_client as http_client
-from urllib.error import URLError
-from typing import Union
+from py_eureka_client import eureka_client, http_client
 
 logger = logging.getLogger(__name__)
 
@@ -16,6 +18,7 @@ class _OAuth2HttpClient(http_client.HttpClient):
     """
 
     def __init__(self, credentials: dict):
+        """Initialize the OAuth2 HTTP client with credentials."""
         self.oauth_client = OAuth2Client(
             client_id=credentials['client_id'],
             client_secret=credentials['client_secret'],
@@ -24,29 +27,32 @@ def __init__(self, credentials: dict):
 
     async def urlopen(self, request: Union[str, http_client.HttpRequest] = None,
                       data: bytes = None, timeout: float = None) -> http_client.HttpResponse:
+        """Open a URL with OAuth2 authentication."""
         if isinstance(request, http_client.HttpRequest):
             req = request
         elif isinstance(request, str):
             req = http_client.HttpRequest(request, headers={'Accept-Encoding': 'gzip'})
         else:
             raise URLError("Invalid URL")
 
-        req = req._to_urllib_request()
+        req = req._to_urllib_request()  # pylint: disable=protected-access
         req.add_header("Connection", "close")
         req.add_header("Authorization", f"Bearer {self.oauth_client.get_access_token()}")
-        res = urllib.request.urlopen(req, data=data, timeout=timeout)
-        return http_client.HttpResponse(res)
+        with urllib.request.urlopen(req, data=data, timeout=timeout) as res:
+            return http_client.HttpResponse(res)
+
+    def get_oauth_client(self):
+        """Get the OAuth2 client instance."""
+        return self.oauth_client
 
 def _on_err(err_type: str, err: Exception):
-    logger.error(f"Eureka client error - Type: {err_type}, Error: {err}")
+    """Handle Eureka client errors."""
+    logger.error("Eureka client error - Type: %s, Error: %s", err_type, err)
     if err_type in (eureka_client.ERROR_REGISTER, eureka_client.ERROR_DISCOVER):
         eureka_client.stop()
 
 def init_client():
-    """
-    Initializes the eureka client.
-    """
-
+    """Initialize the eureka client."""
     env = AppEnv()
 
     service_registry = env.get_service(label="p.service-registry")
@@ -58,16 +64,16 @@ def init_client():
     http_client.set_http_client(_OAuth2HttpClient(credentials))
 
     eureka_client.init(
-                app_name=env.name,
-                instance_host=env.uris[0],
-                instance_ip=os.getenv('CF_INSTANCE_INTERNAL_IP'),
-                instance_id=f"{env.uris[0]}:{env.app.get('instance_id') or '0'}",
-                instance_unsecure_port_enabled=True,
-                instance_port=80,
-                instance_secure_port_enabled=True,
-                instance_secure_port=443,
-                vip_adr='UNKNOWN',
-                secure_vip_addr='UNKNOWN',
-                eureka_server=credentials['uri'],
-                on_error=_on_err
-            )
+        app_name=env.name,
+        instance_host=env.uris[0],
+        instance_ip=os.getenv('CF_INSTANCE_INTERNAL_IP'),
+        instance_id=f"{env.uris[0]}:{env.app.get('instance_id') or '0'}",
+        instance_unsecure_port_enabled=True,
+        instance_port=80,
+        instance_secure_port_enabled=True,
+        instance_secure_port=443,
+        vip_adr='UNKNOWN',
+        secure_vip_addr='UNKNOWN',
+        eureka_server=credentials['uri'],
+        on_error=_on_err
+    )

greeter-messages/oauth2.py

@@ -1,3 +1,5 @@
+"""OAuth2 client for handling client credentials flow with token caching."""
+
 import time
 import logging
 from typing import Optional
@@ -6,25 +8,26 @@
 logger = logging.getLogger(__name__)
 
 class OAuth2Client:
-    """OAuth 2.0 client for handling client credentials flow with token caching"""
+    """OAuth 2.0 client for handling client credentials flow with token caching."""
 
     def __init__(self, client_id: str, client_secret: str, access_token_uri: str):
+        """Initialize the OAuth2 client."""
         self.client_id = client_id
         self.client_secret = client_secret
         self.access_token_uri = access_token_uri
         self.access_token: Optional[str] = None
         self.token_expires_at: Optional[float] = None
 
     def _is_token_expired(self, buffer_seconds: int = 30) -> bool:
-        """Check if the current token is expired or will expire soon"""
+        """Check if the current token is expired or will expire soon."""
         if not self.access_token or not self.token_expires_at:
             return True
 
         # Add buffer time to refresh token before it actually expires
         return time.time() >= (self.token_expires_at - buffer_seconds)
 
     def _request_new_token(self) -> str:
-        """Request a new access token from the OAuth server"""
+        """Request a new access token from the OAuth server."""
         # Prepare token request
         token_data = {
             'grant_type': 'client_credentials',
@@ -58,13 +61,18 @@ def _request_new_token(self) -> str:
 
             return self.access_token
         except requests.exceptions.RequestException as e:
-            logger.error(f"Failed to get access token: {e}")
+            logger.error("Failed to get access token: %s", e)
             raise ValueError(f"Failed to get access token: {e}") from e
 
     def get_access_token(self) -> str:
-        """Get OAuth 2.0 access token using client credentials flow"""
+        """Get OAuth 2.0 access token using client credentials flow."""
         # Check if we have a valid cached token
         if not self._is_token_expired():
             return self.access_token
         # Token is expired or doesn't exist, get a new one
         return self._request_new_token()
+
+    def clear_cache(self) -> None:
+        """Clear the cached token."""
+        self.access_token = None
+        self.token_expires_at = None

greeter/app.py

@@ -1,3 +1,5 @@
+"""Flask application for the greeter service."""
+
 import os
 import logging
 from flask import Flask, request
@@ -11,16 +13,17 @@
 try:
     init_client()
     logger.info("Eureka client initialized successfully")
-except Exception as e:
-    logger.error(f"Failed to initialize Eureka client: {e}")
+except (ValueError, ImportError, OSError) as e:
+    logger.error("Failed to initialize Eureka client: %s", e)
 
 @app.route('/hello')
 def hello():
+    """Handle hello requests by calling the greeter-messages service."""
     salutation = request.args.get('salutation', 'Hello')
     name = request.args.get('name', 'Bob')
-    
+
     return call_service("GREETER-MESSAGES", f"/greeting?salutation={salutation}&name={name}")
 
 if __name__ == '__main__':
     port = int(os.getenv('PORT', '8080'))
-    app.run(host='0.0.0.0', port=port)
+    app.run(host='0.0.0.0', port=port)

greeter/eureka.py

@@ -1,12 +1,14 @@
+"""Eureka client integration for service discovery and registration."""
+
 import os
 import urllib.request
 import logging
+from typing import Union
+from urllib.error import URLError
+
 from cfenv import AppEnv
 from oauth2 import OAuth2Client
-import py_eureka_client.eureka_client as eureka_client
-import py_eureka_client.http_client as http_client
-from urllib.error import URLError
-from typing import Union
+from py_eureka_client import eureka_client, http_client
 
 logger = logging.getLogger(__name__)
 
@@ -16,6 +18,7 @@ class _OAuth2HttpClient(http_client.HttpClient):
     """
 
     def __init__(self, credentials: dict):
+        """Initialize the OAuth2 HTTP client with credentials."""
         self.oauth_client = OAuth2Client(
             client_id=credentials['client_id'],
             client_secret=credentials['client_secret'],
@@ -24,29 +27,32 @@ def __init__(self, credentials: dict):
 
     async def urlopen(self, request: Union[str, http_client.HttpRequest] = None,
                       data: bytes = None, timeout: float = None) -> http_client.HttpResponse:
+        """Open a URL with OAuth2 authentication."""
         if isinstance(request, http_client.HttpRequest):
             req = request
         elif isinstance(request, str):
             req = http_client.HttpRequest(request, headers={'Accept-Encoding': 'gzip'})
         else:
             raise URLError("Invalid URL")
 
-        req = req._to_urllib_request()
+        req = req._to_urllib_request()  # pylint: disable=protected-access
         req.add_header("Connection", "close")
         req.add_header("Authorization", f"Bearer {self.oauth_client.get_access_token()}")
-        res = urllib.request.urlopen(req, data=data, timeout=timeout)
-        return http_client.HttpResponse(res)
+        with urllib.request.urlopen(req, data=data, timeout=timeout) as res:
+            return http_client.HttpResponse(res)
+
+    def get_oauth_client(self):
+        """Get the OAuth2 client instance."""
+        return self.oauth_client
 
 def _on_err(err_type: str, err: Exception):
-    logger.error(f"Eureka client error - Type: {err_type}, Error: {err}")
+    """Handle Eureka client errors."""
+    logger.error("Eureka client error - Type: %s, Error: %s", err_type, err)
     if err_type in (eureka_client.ERROR_REGISTER, eureka_client.ERROR_DISCOVER):
         eureka_client.stop()
 
 def init_client():
-    """
-    Initializes the eureka client.
-    """
-
+    """Initialize the eureka client."""
     env = AppEnv()
 
     service_registry = env.get_service(label="p.service-registry")
@@ -58,26 +64,24 @@ def init_client():
     http_client.set_http_client(_OAuth2HttpClient(credentials))
 
     eureka_client.init(
-                app_name=env.name,
-                instance_host=env.uris[0],
-                instance_ip=os.getenv('CF_INSTANCE_INTERNAL_IP'),
-                instance_id=f"{env.uris[0]}:{env.app.get('instance_id') or '0'}",
-                instance_unsecure_port_enabled=True,
-                instance_port=80,
-                instance_secure_port_enabled=True,
-                instance_secure_port=443,
-                vip_adr='UNKNOWN',
-                secure_vip_addr='UNKNOWN',
-                eureka_server=credentials['uri'],
-                on_error=_on_err
-            )
+        app_name=env.name,
+        instance_host=env.uris[0],
+        instance_ip=os.getenv('CF_INSTANCE_INTERNAL_IP'),
+        instance_id=f"{env.uris[0]}:{env.app.get('instance_id') or '0'}",
+        instance_unsecure_port_enabled=True,
+        instance_port=80,
+        instance_secure_port_enabled=True,
+        instance_secure_port=443,
+        vip_adr='UNKNOWN',
+        secure_vip_addr='UNKNOWN',
+        eureka_server=credentials['uri'],
+        on_error=_on_err
+    )
 
 def call_service(name: str, path: str):
-    """
-    Calls a service with the given name and path.
-    """
+    """Call a service with the given name and path."""
     try:
         return eureka_client.do_service(app_name=name, service=path, prefer_https=True)
     except urllib.request.HTTPError as e:
-        logger.error(f"Error calling service {name} at {path}: {e}")
-        raise e
+        logger.error("Error calling service %s at %s: %s", name, path, e)
+        raise e

greeter/oauth2.py

@@ -1,3 +1,5 @@
+"""OAuth2 client for handling client credentials flow with token caching."""
+
 import time
 import logging
 from typing import Optional
@@ -6,25 +8,26 @@
 logger = logging.getLogger(__name__)
 
 class OAuth2Client:
-    """OAuth 2.0 client for handling client credentials flow with token caching"""
+    """OAuth 2.0 client for handling client credentials flow with token caching."""
 
     def __init__(self, client_id: str, client_secret: str, access_token_uri: str):
+        """Initialize the OAuth2 client."""
         self.client_id = client_id
         self.client_secret = client_secret
         self.access_token_uri = access_token_uri
         self.access_token: Optional[str] = None
         self.token_expires_at: Optional[float] = None
 
     def _is_token_expired(self, buffer_seconds: int = 30) -> bool:
-        """Check if the current token is expired or will expire soon"""
+        """Check if the current token is expired or will expire soon."""
         if not self.access_token or not self.token_expires_at:
             return True
 
         # Add buffer time to refresh token before it actually expires
         return time.time() >= (self.token_expires_at - buffer_seconds)
 
     def _request_new_token(self) -> str:
-        """Request a new access token from the OAuth server"""
+        """Request a new access token from the OAuth server."""
         # Prepare token request
         token_data = {
             'grant_type': 'client_credentials',
@@ -58,13 +61,18 @@ def _request_new_token(self) -> str:
 
             return self.access_token
         except requests.exceptions.RequestException as e:
-            logger.error(f"Failed to get access token: {e}")
+            logger.error("Failed to get access token: %s", e)
             raise ValueError(f"Failed to get access token: {e}") from e
 
     def get_access_token(self) -> str:
-        """Get OAuth 2.0 access token using client credentials flow"""
+        """Get OAuth 2.0 access token using client credentials flow."""
         # Check if we have a valid cached token
         if not self._is_token_expired():
             return self.access_token
         # Token is expired or doesn't exist, get a new one
         return self._request_new_token()
+
+    def clear_cache(self) -> None:
+        """Clear the cached token."""
+        self.access_token = None
+        self.token_expires_at = None