config_server.py

import os
from typing import Dict, Any
from cfenv import AppEnv
import requests
from oauth2 import OAuth2Client

class ConfigServerClient:
    def __init__(self):
        self.env = AppEnv()
        self.oauth_client = None
        self.credentials = None

        # Get configuration parameters from environment variables
        self.name = os.getenv('APPLICATION_NAME', self.env.name)
        self.profiles = os.getenv('PROFILES', 'default')
        self.label = os.getenv('LABEL', 'main')

        config_server = self.env.get_service(label="p.config-server")
        if not config_server:
            raise ValueError("No 'p.config-server' service instance found.")

        self.credentials = config_server.credentials
        self.oauth_client = OAuth2Client(
            client_id=self.credentials['client_id'],
            client_secret=self.credentials['client_secret'],
            access_token_uri=self.credentials['access_token_uri']
        )

    def _make_http_request(self, url: str, accept_type: str) -> requests.Response:
        """Make an HTTP request to the config server with the specified accept type"""
        headers = {
            'Authorization': f"Bearer {self.oauth_client.get_access_token()}",
            'Accept': accept_type
        }

        try:
            response = requests.get(url, headers=headers, timeout=30)
            response.raise_for_status()
            return response
        except requests.exceptions.RequestException as e:
            raise e

    def load(self) -> Dict[str, Any]:
        """Load and merge configuration properties from the config server"""
        url = f"{self.credentials['uri']}/{self.name}/{self.profiles}/{self.label}"

        response = self._make_http_request(url, 'application/json')

        config = {}
        property_sources = response.json().get('propertySources', [])

        for property_source in property_sources:
            config.update(property_source.get('source', {}))

        return config

    def load_text_resource(self, name: str) -> str:
        """Load a text resource from the config server"""
        url = f"{self.credentials['uri']}/{self.name}/{self.profiles}/{self.label}/{name}"

        response = self._make_http_request(url, 'text/plain')
        return response.text

    def load_binary_resource(self, name: str) -> bytes:
        """Load a binary resource from the config server"""
        url = f"{self.credentials['uri']}/{self.name}/{self.profiles}/{self.label}/{name}"

        response = self._make_http_request(url, 'application/octet-stream')
        return response.content