| 123456789101112131415161718192021222324252627282930 |
- import json
- from typing import Optional, Dict
- from applications.utils import AsyncApolloClient
- class AsyncApolloApi:
- def __init__(self, app_id="LongArticlesJob", env="pre"):
- match env:
- case "pre":
- config_server_url = "http://preapolloconfig-internal.piaoquantv.com/"
- case "dev":
- config_server_url = "https://devapolloconfig-internal.piaoquantv.com/"
- case "prod":
- config_server_url = "https://apolloconfig-internal.piaoquantv.com/"
- case _:
- raise ValueError("env must be 'pre' or 'dev' or 'prod'")
- self.apollo_connection = AsyncApolloClient(
- app_id=app_id, config_server_url=config_server_url
- )
- async def get_config_value(
- self, key: str, output_type: str = "json"
- ) -> Optional[Dict]:
- match output_type:
- case "json":
- response = await self.apollo_connection.get_value(key)
- return json.loads(response)
- case _:
- return await self.apollo_connection.get_value(key)
|
