admin/Azure-co-op-translator
1
0
Fork 0
mirror of https://github.com/Azure/co-op-translator synced 2026-08-09 12:00:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Core: Add full multi-environment-set support for OpenAI, Azure OpenAI, and Azure Vision providers (#307)

Browse Source
This commit is contained in:
Minseok Song
2025-12-13 22:54:22 +09:00
committed by GitHub
parent 171790eb4d
commit b17bd90433
13 changed files with 409 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.template
+9 -1
View File
@@ -2,6 +2,10 @@
AZURE_AI_SERVICE_API_KEY="your_azure_ai_service_api_key" # API key from Azure AI Service resource
AZURE_AI_SERVICE_ENDPOINT="https://your_azure_ai_service_endpoint" # Endpoint from Azure AI Service resource
# Optional fallback set example (index 1)
AZURE_AI_SERVICE_API_KEY_1="your_azure_ai_service_api_key_1"
AZURE_AI_SERVICE_ENDPOINT_1="https://your_azure_ai_service_endpoint_1"
# Azure OpenAI Credentials (Required for text translation)
AZURE_OPENAI_API_KEY="your_azure_openai_api_key"
AZURE_OPENAI_ENDPOINT="https://your_azure_openai_endpoint"
@@ -9,8 +13,12 @@ AZURE_OPENAI_MODEL_NAME="your_model_name"
AZURE_OPENAI_CHAT_DEPLOYMENT_NAME="your_deployment_name"
AZURE_OPENAI_API_VERSION="your_api_version"
# Optional fallback sets: duplicate the full AZURE_OPENAI_* set with suffix _1/_2 (same index for all variables)
# OpenAI Credentials
OPENAI_API_KEY="your_openai_api_key"
OPENAI_ORG_ID="your_openai_org_id"
OPENAI_CHAT_MODEL_ID="your_chat_model_id(ex. gpt-4o)"
OPENAI_BASE_URL="https://api.openai.com/v1 (If you don't have a custom base URL, you can delete this line, then it will use the default base URL)"
OPENAI_BASE_URL="https://api.openai.com/v1 (If you don't have a custom base URL, you can delete this line, then it will use the default base URL)"
# Optional fallback sets: duplicate the full OPENAI_* set with suffix _1/_2 (same index for all variables)
README.md
+4 -3
View File
@@ -70,9 +70,10 @@ docker run --rm -it --env-file .env -v "${PWD}:/work" ghcr.io/azure/co-op-transl
2. Create a `.env` file using the template: [.env.template](./.env.template)
3. Configure one LLM provider (Azure OpenAI or OpenAI)
4. (Optional) For image translation (`-img`), configure Azure AI Vision
5. (Recommended) Clean up any previous translations to avoid conflicts (e.g., `translations/`)
6. (Recommended) Add a translation section to your README using the [README languages template](./getting_started/README_languages_template.md)
7. See: [Set up Azure AI](./getting_started/set-up-azure-ai.md)
5. (Optional) You can configure multiple credential sets by duplicating variables with suffixes like `_1`, `_2`, etc. All variables in a set must share the same suffix.
6. (Recommended) Clean up any previous translations to avoid conflicts (e.g., `translations/`)
7. (Recommended) Add a translation section to your README using the [README languages template](./getting_started/README_languages_template.md)
8. See: [Set up Azure AI](./getting_started/set-up-azure-ai.md)
## Usage
getting_started/command-line-guide/create-env-file.md
+8
View File
@@ -27,6 +27,10 @@ In the root directory of your project, create a file named *.env*. This file wil
AZURE_AI_SERVICE_API_KEY="your_azure_ai_service_api_key"
AZURE_AI_SERVICE_ENDPOINT="https://your_azure_ai_service_endpoint"
# Optional fallback set example (index 1)
AZURE_AI_SERVICE_API_KEY_1="your_azure_ai_service_api_key_1"
AZURE_AI_SERVICE_ENDPOINT_1="https://your_azure_ai_service_endpoint_1"
# Azure OpenAI Credentials
AZURE_OPENAI_API_KEY="your_azure_openai_api_key"
AZURE_OPENAI_ENDPOINT="https://your_azure_openai_endpoint"
@@ -34,11 +38,15 @@ In the root directory of your project, create a file named *.env*. This file wil
AZURE_OPENAI_CHAT_DEPLOYMENT_NAME="your_deployment_name"
AZURE_OPENAI_API_VERSION="your_api_version"
# Optional fallback sets: duplicate the full AZURE_OPENAI_* set with suffix _1/_2 (same index for all variables)
# OpenAI Credentials
OPENAI_API_KEY="your_openai_api_key"
OPENAI_ORG_ID="your_openai_org_id"
OPENAI_CHAT_MODEL_ID="your_chat_model_id(ex. gpt-4o)"
OPENAI_BASE_URL="https://api.openai.com/v1 (If you don't have a custom base URL, you can delete this lin, then it will use the default base URL)"
# Optional fallback sets: duplicate the full OPENAI_* set with suffix _1/_2 (same index for all variables)
```
> [!NOTE]
getting_started/set-up-azure-ai.md
+6
View File
@@ -91,12 +91,18 @@ Later, when building your application, you'll likely configure it using these co
AZURE_AI_SERVICE_API_KEY="your_azure_ai_service_api_key" # e.g., 21xasd...
AZURE_AI_SERVICE_ENDPOINT="https://your_azure_ai_service_endpoint.cognitiveservices.azure.com/"
# Optional fallback sets: duplicate variables with suffix _1/_2 (same index for all variables in the set)
AZURE_AI_SERVICE_API_KEY_1="your_azure_ai_service_api_key_1"
AZURE_AI_SERVICE_ENDPOINT_1="https://your_azure_ai_service_endpoint_1.cognitiveservices.azure.com/"
# Azure OpenAI Credentials (Required for text translation)
AZURE_OPENAI_API_KEY="your_azure_openai_api_key" # e.g., 21xasd...
AZURE_OPENAI_ENDPOINT="https://your_azure_openai_endpoint.openai.azure.com/"
AZURE_OPENAI_MODEL_NAME="your_model_name" # e.g., gpt-4o
AZURE_OPENAI_CHAT_DEPLOYMENT_NAME="your_deployment_name" # e.g., cooptranslator-gpt4o
AZURE_OPENAI_API_VERSION="your_api_version" # e.g., 2024-12-01-preview
# Optional fallback sets: duplicate the full AZURE_OPENAI_* set with suffix _1/_2 (same index for all variables)
```
---
pyproject.toml
+1 -1
View File
@@ -88,7 +88,7 @@ urllib3 = "^2.2.1"
wcwidth = "^0.2.13"
arabic-reshaper = "^3.0.0"
python-bidi = "^0.6.6"
az-ai-healthcheck = { version = "^0.1.5", extras = ["vision"] }
az-ai-healthcheck = { version = "^0.1.6", extras = ["vision"] }
ai-healthcheck = "^0.1.1"
markdown-it-py = "^3.0.0"
src/co_op_translator/config/llm_config/azure_openai.py
+39 -5
View File
@@ -1,6 +1,8 @@
import os
from dotenv import load_dotenv
from co_op_translator.utils.common.env_set_utils import get_active_env_set, get_env_sets
# Load environment variables from .env file
load_dotenv()
@@ -8,27 +10,59 @@ load_dotenv()
class AzureOpenAIConfig:
"""Azure OpenAI specific configuration."""
_GROUP = "azure_openai"
_REQUIRED = (
"AZURE_OPENAI_API_KEY",
"AZURE_OPENAI_ENDPOINT",
"AZURE_OPENAI_MODEL_NAME",
"AZURE_OPENAI_CHAT_DEPLOYMENT_NAME",
"AZURE_OPENAI_API_VERSION",
)
@staticmethod
def get_env_sets():
return get_env_sets(group=AzureOpenAIConfig._GROUP, required=AzureOpenAIConfig._REQUIRED)
@staticmethod
def get_active_env_set():
return get_active_env_set(group=AzureOpenAIConfig._GROUP, required=AzureOpenAIConfig._REQUIRED)
@staticmethod
def get_api_key():
"""Retrieve the Azure OpenAI API key from environment variables."""
return os.getenv("AZURE_OPENAI_API_KEY")
env_set = AzureOpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_OPENAI_API_KEY")
@staticmethod
def get_endpoint():
"""Retrieve the Azure OpenAI endpoint from environment variables."""
return os.getenv("AZURE_OPENAI_ENDPOINT")
env_set = AzureOpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_OPENAI_ENDPOINT")
@staticmethod
def get_model_name():
"""Retrieve the Azure OpenAI model name from environment variables."""
return os.getenv("AZURE_OPENAI_MODEL_NAME")
env_set = AzureOpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_OPENAI_MODEL_NAME")
@staticmethod
def get_chat_deployment_name():
"""Retrieve the Azure OpenAI chat deployment name from environment variables."""
return os.getenv("AZURE_OPENAI_CHAT_DEPLOYMENT_NAME")
env_set = AzureOpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_OPENAI_CHAT_DEPLOYMENT_NAME")
@staticmethod
def get_api_version():
"""Retrieve the Azure OpenAI API version from environment variables."""
return os.getenv("AZURE_OPENAI_API_VERSION")
env_set = AzureOpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_OPENAI_API_VERSION")
src/co_op_translator/config/llm_config/config.py
+69 -46
View File
@@ -8,6 +8,7 @@ from az_ai_healthcheck import check_azure_openai
from co_op_translator.config.llm_config.provider import LLMProvider
from co_op_translator.config.llm_config.azure_openai import AzureOpenAIConfig
from co_op_translator.config.llm_config.openai import OpenAIConfig
from co_op_translator.utils.common.env_set_utils import any_env_var_present, set_preferred_env_set
logger = logging.getLogger(__name__)
@@ -36,26 +37,27 @@ class LLMConfig:
- "NO_CONFIG" if no variables are set at all.
- "Incomplete" if some are set but not all (or required ones are missing).
"""
non_empty_count = sum(bool(v) for v in env_vars.values())
if provider == LLMProvider.OPENAI:
# If there's no environment variable filled at all
if non_empty_count == 0:
bases = ["OPENAI_API_KEY", "OPENAI_CHAT_MODEL_ID", "OPENAI_ORG_ID", "OPENAI_BASE_URL"]
if not any_env_var_present(bases):
raise ValueError("NO_CONFIG_OPENAI")
# If OPENAI_API_KEY is missing or empty, it's incomplete
if not env_vars.get("OPENAI_API_KEY"):
raise ValueError(
"Incomplete OpenAI configuration. The 'OPENAI_API_KEY' must be set."
)
if not env_vars.get("OPENAI_CHAT_MODEL_ID"):
raise ValueError(
"Incomplete OpenAI configuration. The 'OPENAI_CHAT_MODEL_ID' must be set."
)
elif provider == LLMProvider.AZURE_OPENAI:
# If there's no environment variable filled at all
if non_empty_count == 0:
bases = list(env_vars.keys())
if not any_env_var_present(bases):
raise ValueError("NO_CONFIG_AZURE")
# If some but not all are filled, it's incomplete
if non_empty_count < len(env_vars):
if any(v is None or not str(v).strip() for v in env_vars.values()):
raise ValueError(
f"Incomplete {provider.name} configuration. Ensure all required environment variables are set."
)
@@ -156,55 +158,76 @@ class LLMConfig:
provider = cls.get_available_provider()
if provider == LLMProvider.AZURE_OPENAI:
endpoint = (AzureOpenAIConfig.get_endpoint() or "").rstrip("/")
api_version = AzureOpenAIConfig.get_api_version()
api_key = AzureOpenAIConfig.get_api_key()
deployment = AzureOpenAIConfig.get_chat_deployment_name()
if not endpoint or not api_version or not api_key or not deployment:
env_sets = AzureOpenAIConfig.get_env_sets()
if not env_sets:
raise ValueError(
"Azure OpenAI configuration missing required values. Ensure AZURE_OPENAI_ENDPOINT, AZURE_OPENAI_API_VERSION, AZURE_OPENAI_API_KEY, and AZURE_OPENAI_CHAT_DEPLOYMENT_NAME are set."
)
# Use external healthcheck helper for minimal probe
last_message: Optional[str] = None
for env_set in env_sets:
endpoint = (env_set.values.get("AZURE_OPENAI_ENDPOINT") or "").rstrip("/")
api_version = env_set.values.get("AZURE_OPENAI_API_VERSION")
api_key = env_set.values.get("AZURE_OPENAI_API_KEY")
deployment = env_set.values.get("AZURE_OPENAI_CHAT_DEPLOYMENT_NAME")
res = check_azure_openai(
endpoint=endpoint,
api_key=api_key,
api_version=api_version,
deployment=deployment,
timeout=10.0,
)
if not endpoint or not api_version or not api_key or not deployment:
continue
if res.ok:
return True
# Fail on any non-ok with package-provided message
raise ValueError(res.message)
try:
res = check_azure_openai(
endpoint=endpoint,
api_key=api_key,
api_version=api_version,
deployment=deployment,
timeout=10.0,
)
except Exception as e:
last_message = str(e)
continue
if res.ok:
set_preferred_env_set(AzureOpenAIConfig._GROUP, env_set.index)
return True
last_message = res.message
raise ValueError(last_message or "Azure OpenAI connectivity check failed")
elif provider == LLMProvider.OPENAI:
api_key = OpenAIConfig.get_api_key()
base_url = OpenAIConfig.get_base_url()
org_id = OpenAIConfig.get_org_id()
model_id = OpenAIConfig.get_chat_model_id()
if not api_key:
raise ValueError("OPENAI_API_KEY must be set for OpenAI provider.")
if not model_id:
env_sets = OpenAIConfig.get_env_sets()
if not env_sets:
raise ValueError(
"OPENAI_CHAT_MODEL_ID must be set for OpenAI provider."
"OpenAI configuration missing required values. Ensure OPENAI_API_KEY and OPENAI_CHAT_MODEL_ID are set."
)
res = check_openai(
endpoint=base_url,
api_key=api_key,
model=model_id,
org_id=org_id,
timeout=10.0,
)
last_message: Optional[str] = None
for env_set in env_sets:
api_key = env_set.values.get("OPENAI_API_KEY")
base_url = env_set.values.get("OPENAI_BASE_URL")
org_id = env_set.values.get("OPENAI_ORG_ID")
model_id = env_set.values.get("OPENAI_CHAT_MODEL_ID")
if res.ok:
return True
raise ValueError(res.message)
if not api_key or not model_id:
continue
try:
res = check_openai(
endpoint=base_url,
api_key=api_key,
model=model_id,
org_id=org_id,
timeout=10.0,
)
except Exception as e:
last_message = str(e)
continue
if res.ok:
set_preferred_env_set(OpenAIConfig._GROUP, env_set.index)
return True
last_message = res.message
raise ValueError(last_message or "OpenAI connectivity check failed")
else:
# Should not happen because get_available_provider() would have raised earlier otherwise
raise ValueError("No LLM provider available for connectivity validation.")
src/co_op_translator/config/llm_config/openai.py
+49 -4
View File
@@ -1,6 +1,8 @@
import os
from dotenv import load_dotenv
from co_op_translator.utils.common.env_set_utils import get_active_env_set, get_env_sets
# Load environment variables from .env file
load_dotenv()
@@ -8,20 +10,60 @@ load_dotenv()
class OpenAIConfig:
"""OpenAI specific configuration."""
_GROUP = "openai"
_REQUIRED = (
"OPENAI_API_KEY",
"OPENAI_CHAT_MODEL_ID",
)
_OPTIONAL = (
"OPENAI_ORG_ID",
"OPENAI_BASE_URL",
)
_DEFAULTS = {
"OPENAI_BASE_URL": "https://api.openai.com/v1",
}
@staticmethod
def get_env_sets():
return get_env_sets(
group=OpenAIConfig._GROUP,
required=OpenAIConfig._REQUIRED,
optional=OpenAIConfig._OPTIONAL,
defaults=OpenAIConfig._DEFAULTS,
)
@staticmethod
def get_active_env_set():
return get_active_env_set(
group=OpenAIConfig._GROUP,
required=OpenAIConfig._REQUIRED,
optional=OpenAIConfig._OPTIONAL,
defaults=OpenAIConfig._DEFAULTS,
)
@staticmethod
def get_api_key():
"""Retrieve the OpenAI API key from environment variables."""
return os.getenv("OPENAI_API_KEY")
env_set = OpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("OPENAI_API_KEY")
@staticmethod
def get_org_id():
"""Retrieve the OpenAI organization ID from environment variables."""
return os.getenv("OPENAI_ORG_ID")
env_set = OpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("OPENAI_ORG_ID")
@staticmethod
def get_chat_model_id():
"""Retrieve the OpenAI chat model ID from environment variables."""
return os.getenv("OPENAI_CHAT_MODEL_ID")
env_set = OpenAIConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("OPENAI_CHAT_MODEL_ID")
@staticmethod
def get_api_version():
@@ -31,4 +73,7 @@ class OpenAIConfig:
@staticmethod
def get_base_url():
"""Retrieve the OpenAI base URL from environment variables or return default."""
return os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
env_set = OpenAIConfig.get_active_env_set()
if env_set is None:
return OpenAIConfig._DEFAULTS["OPENAI_BASE_URL"]
return env_set.values.get("OPENAI_BASE_URL", OpenAIConfig._DEFAULTS["OPENAI_BASE_URL"])
src/co_op_translator/config/vision_config/azure_computer_vision.py
+29 -10
View File
@@ -1,25 +1,44 @@
import os
from co_op_translator.utils.common.env_set_utils import get_active_env_set, get_env_sets
class AzureAIVisionConfig:
"""Azure AI Service specific configuration."""
_GROUP = "azure_ai_service"
_REQUIRED = (
"AZURE_AI_SERVICE_API_KEY",
"AZURE_AI_SERVICE_ENDPOINT",
)
@staticmethod
def get_env_sets():
return get_env_sets(group=AzureAIVisionConfig._GROUP, required=AzureAIVisionConfig._REQUIRED)
@staticmethod
def get_active_env_set():
return get_active_env_set(
group=AzureAIVisionConfig._GROUP,
required=AzureAIVisionConfig._REQUIRED,
)
@staticmethod
def get_api_key():
"""Retrieve the Azure AI Service API key from environment variables.
First checks for AZURE_AI_SERVICE_API_KEY (recommended), then falls back to
AZURE_SUBSCRIPTION_KEY for backward compatibility.
First checks for AZURE_AI_SERVICE_API_KEY (recommended), then checks numbered
variants like AZURE_AI_SERVICE_API_KEY_1.
"""
# First check new naming convention (Azure AI Service)
key = os.getenv("AZURE_AI_SERVICE_API_KEY")
if key:
return key
# Fall back to legacy naming convention for backward compatibility
return os.getenv("AZURE_SUBSCRIPTION_KEY")
env_set = AzureAIVisionConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_AI_SERVICE_API_KEY")
@staticmethod
def get_endpoint():
"""Retrieve the Azure AI service endpoint from environment variables."""
return os.getenv("AZURE_AI_SERVICE_ENDPOINT")
env_set = AzureAIVisionConfig.get_active_env_set()
if env_set is None:
return None
return env_set.values.get("AZURE_AI_SERVICE_ENDPOINT")
src/co_op_translator/config/vision_config/config.py
+25 -13
View File
@@ -6,6 +6,7 @@ from co_op_translator.config.vision_config.azure_computer_vision import (
AzureAIVisionConfig,
)
from az_ai_healthcheck import check_azure_ai_vision
from co_op_translator.utils.common.env_set_utils import set_preferred_env_set
logger = logging.getLogger(__name__)
@@ -90,21 +91,32 @@ class VisionConfig:
if provider != VisionProvider.AZURE_COMPUTER_VISION:
return
endpoint = AzureAIVisionConfig.get_endpoint()
api_key = AzureAIVisionConfig.get_api_key()
if not endpoint or not api_key:
env_sets = AzureAIVisionConfig.get_env_sets()
if not env_sets:
raise ValueError(
"Azure AI Service configuration missing required values. Ensure AZURE_AI_SERVICE_ENDPOINT and AZURE_AI_SERVICE_API_KEY are set."
)
# Use healthcheck helper with default 50x50 in-memory PNG
res = check_azure_ai_vision(
endpoint=endpoint,
api_key=api_key,
timeout=10.0,
)
last_message: Optional[str] = None
for env_set in env_sets:
endpoint = env_set.values.get("AZURE_AI_SERVICE_ENDPOINT")
api_key = env_set.values.get("AZURE_AI_SERVICE_API_KEY")
if not endpoint or not api_key:
continue
if res.ok:
return True
# Fail on any non-ok with package-provided message
raise ValueError(res.message)
try:
res = check_azure_ai_vision(
endpoint=endpoint,
api_key=api_key,
timeout=10.0,
)
except Exception as e:
last_message = str(e)
continue
if res.ok:
set_preferred_env_set(AzureAIVisionConfig._GROUP, env_set.index)
return True
last_message = res.message
raise ValueError(last_message or "Azure AI Vision connectivity check failed")
src/co_op_translator/core/vision/providers/azure/image_translator.py
+38
View File
@@ -1,10 +1,12 @@
import logging
from azure.ai.vision.imageanalysis import ImageAnalysisClient
from azure.core.exceptions import HttpResponseError, ServiceRequestError
from azure.core.credentials import AzureKeyCredential
from co_op_translator.core.vision.image_translator import ImageTranslator
from co_op_translator.config.vision_config.azure_computer_vision import (
AzureAIVisionConfig,
)
from co_op_translator.utils.common.env_set_utils import set_preferred_env_set
logger = logging.getLogger(__name__)
@@ -29,3 +31,39 @@ class AzureImageTranslator(ImageTranslator):
endpoint = AzureAIVisionConfig.get_endpoint()
subscription_key = AzureAIVisionConfig.get_api_key()
return ImageAnalysisClient(endpoint, AzureKeyCredential(subscription_key))
def extract_line_bounding_boxes(self, image_path):
env_sets = AzureAIVisionConfig.get_env_sets()
if not env_sets:
return super().extract_line_bounding_boxes(image_path)
last_exc = None
for env_set in env_sets:
set_preferred_env_set(AzureAIVisionConfig._GROUP, env_set.index)
try:
result = super().extract_line_bounding_boxes(image_path)
return result
except HttpResponseError as e:
status_code = getattr(e, "status_code", None)
if status_code is None and getattr(e, "response", None) is not None:
status_code = getattr(e.response, "status_code", None)
last_exc = e
if status_code in {401, 403, 408, 429, 500, 502, 503, 504}:
logger.warning(
"Azure AI Vision request failed (status=%s); trying next env set",
status_code,
)
continue
raise
except ServiceRequestError as e:
last_exc = e
logger.warning(
"Azure AI Vision request error; trying next env set: %s", str(e)
)
continue
if last_exc is not None:
raise last_exc
return super().extract_line_bounding_boxes(image_path)
src/co_op_translator/utils/common/env_set_utils.py
+127
View File
@@ -0,0 +1,127 @@
import os
from dataclasses import dataclass
from typing import Dict, List, Optional, Sequence
@dataclass(frozen=True)
class EnvSet:
group: str
index: int
values: Dict[str, str]
_preferred_env_set_index: Dict[str, int] = {}
def set_preferred_env_set(group: str, index: int) -> None:
_preferred_env_set_index[group] = index
def get_preferred_env_set(group: str) -> Optional[int]:
return _preferred_env_set_index.get(group)
def _env_name(base: str, index: int) -> str:
return base if index == 0 else f"{base}_{index}"
def _get_env(base: str, index: int) -> Optional[str]:
value = os.getenv(_env_name(base, index))
if value is None:
return None
value = value.strip()
return value or None
def any_env_var_present(bases: Sequence[str]) -> bool:
for base in bases:
v = os.getenv(base)
if v is not None and v.strip():
return True
for name, value in os.environ.items():
if value is None or not str(value).strip():
continue
for base in bases:
prefix = f"{base}_"
if not name.startswith(prefix):
continue
suffix = name[len(prefix) :]
if suffix.isdigit():
return True
return False
def get_env_sets(
group: str,
required: Sequence[str],
optional: Sequence[str] = (),
defaults: Optional[Dict[str, str]] = None,
) -> List[EnvSet]:
defaults = defaults or {}
indices: set[int] = set()
for base in list(required) + list(optional):
if os.getenv(base):
indices.add(0)
for name in os.environ.keys():
for base in list(required) + list(optional):
prefix = f"{base}_"
if not name.startswith(prefix):
continue
suffix = name[len(prefix) :]
if suffix.isdigit():
indices.add(int(suffix))
ordered_indices = sorted(indices)
preferred = get_preferred_env_set(group)
if preferred is not None and preferred in ordered_indices:
ordered_indices.remove(preferred)
ordered_indices.insert(0, preferred)
env_sets: List[EnvSet] = []
for idx in ordered_indices:
values: Dict[str, str] = {}
ok = True
for base in required:
v = _get_env(base, idx)
if v is None:
ok = False
break
values[base] = v
if not ok:
continue
for base in optional:
v = _get_env(base, idx)
if v is None:
v = defaults.get(base)
if v is not None:
values[base] = v
env_sets.append(EnvSet(group=group, index=idx, values=values))
return env_sets
def get_active_env_set(
group: str,
required: Sequence[str],
optional: Sequence[str] = (),
defaults: Optional[Dict[str, str]] = None,
) -> Optional[EnvSet]:
sets = get_env_sets(group=group, required=required, optional=optional, defaults=defaults)
if not sets:
return None
active = sets[0]
if get_preferred_env_set(group) is None:
set_preferred_env_set(group, active.index)
return active
tests/co_op_translator/config/test_base_config.py
+5 -2
View File
@@ -20,13 +20,16 @@ def azure_openai_env_vars():
@pytest.fixture
def openai_env_vars():
return {"OPENAI_API_KEY": "fake_openai_key"}
return {
"OPENAI_API_KEY": "fake_openai_key",
"OPENAI_CHAT_MODEL_ID": "gpt-4o",
}
@pytest.fixture
def vision_env_vars():
return {
"AZURE_SUBSCRIPTION_KEY": "fake_subscription_key",
"AZURE_AI_SERVICE_API_KEY": "fake_ai_service_key",
"AZURE_AI_SERVICE_ENDPOINT": "https://fake-ai-service-endpoint.com",
}