Fix env config to variables #1790
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,154 @@ | ||
| """Centralized configuration for environment variables.""" | ||
|
|
||
| import os | ||
| from typing import Optional | ||
|
|
||
|
|
||
| class Config: # pylint: disable=too-many-public-methods | ||
| """Centralized configuration for environment variables. | ||
|
|
||
| This class provides type-safe access to environment variables with | ||
| consistent default values across the codebase. | ||
| """ | ||
|
|
||
| # Telemetry/Tracing Configuration | ||
| @classmethod | ||
| def ot_program_name(cls) -> str: | ||
| """Get OpenTelemetry program name.""" | ||
| return os.environ.get("OT_PROGRAM_NAME", "unnamed_execution") | ||
|
|
||
| @classmethod | ||
| def ot_jaeger_host(cls) -> Optional[str]: | ||
| """Get Jaeger host for tracing.""" | ||
| return os.environ.get("OT_JAEGER_HOST_KEY", None) | ||
|
|
||
| @classmethod | ||
| def ot_jaeger_port(cls) -> int: | ||
| """Get Jaeger port for tracing.""" | ||
| return int(os.environ.get("OT_JAEGER_PORT_KEY", "6831")) | ||
|
|
||
| @classmethod | ||
| def ot_traceparent_id(cls) -> Optional[str]: | ||
| """Get OpenTelemetry traceparent ID.""" | ||
| return os.environ.get("OT_TRACEPARENT_ID_KEY", None) | ||
|
|
||
| @classmethod | ||
| def ot_insecure(cls) -> bool: | ||
| """Check if OpenTelemetry should use insecure connection.""" | ||
| return bool(int(os.environ.get("OT_INSECURE", "0"))) | ||
|
|
||
| @classmethod | ||
| def ot_enabled(cls) -> bool: | ||
| """Check if OpenTelemetry tracing is enabled.""" | ||
| return bool(int(os.environ.get("OT_ENABLED", "0"))) | ||
|
|
||
| @classmethod | ||
| def ot_ray_tracer(cls) -> bool: | ||
| """Check if Ray tracer is enabled.""" | ||
| return bool(int(os.environ.get("OT_RAY_TRACER", "0"))) | ||
|
|
||
| # Gateway Configuration | ||
| @classmethod | ||
| def gateway_host(cls) -> Optional[str]: | ||
| """Get gateway provider host.""" | ||
| return os.environ.get("ENV_GATEWAY_PROVIDER_HOST", None) | ||
|
|
||
| @classmethod | ||
| def gateway_version(cls) -> str: | ||
| """Get gateway provider version.""" | ||
| return os.environ.get("ENV_GATEWAY_PROVIDER_VERSION", "v1") | ||
|
|
||
| @classmethod | ||
| def gateway_provider_version(cls) -> str: | ||
| """Get gateway provider version (alias for gateway_version).""" | ||
| return cls.gateway_version() | ||
|
|
||
| @classmethod | ||
| def gateway_token(cls) -> Optional[str]: | ||
| """Get gateway provider token.""" | ||
| return os.environ.get("ENV_GATEWAY_PROVIDER_TOKEN", None) | ||
|
Contributor
There was a problem hiding this comment. Why was the "provider" part removed from the gateway host, token and version? I mean, why not gateway_provider_token, gateway_provider_version and gateway_provider_host instead... is the "provider" word redundant in the gateway? |
||
|
|
||
| # Job Configuration | ||
| @classmethod | ||
| def job_gateway_token(cls) -> Optional[str]: | ||
| """Get job gateway token.""" | ||
| return os.environ.get("ENV_JOB_GATEWAY_TOKEN", None) | ||
|
|
||
| @classmethod | ||
| def job_gateway_instance(cls) -> Optional[str]: | ||
| """Get job gateway instance.""" | ||
| return os.environ.get("ENV_JOB_GATEWAY_INSTANCE", None) | ||
|
|
||
| @classmethod | ||
| def job_gateway_host(cls) -> Optional[str]: | ||
| """Get job gateway host.""" | ||
| return os.environ.get("ENV_JOB_GATEWAY_HOST", None) | ||
|
|
||
| @classmethod | ||
| def job_id_gateway(cls) -> Optional[str]: | ||
| """Get job ID from gateway.""" | ||
| return os.environ.get("ENV_JOB_ID_GATEWAY", None) | ||
|
|
||
| @classmethod | ||
| def qiskit_ibm_channel(cls) -> Optional[str]: | ||
| """Get Qiskit IBM channel.""" | ||
| return os.environ.get("QISKIT_IBM_CHANNEL", None) | ||
|
|
||
| @classmethod | ||
| def qiskit_ibm_token(cls) -> Optional[str]: | ||
| """Get Qiskit IBM token.""" | ||
| return os.environ.get("QISKIT_IBM_TOKEN", None) | ||
|
|
||
| @classmethod | ||
| def qiskit_ibm_url(cls) -> Optional[str]: | ||
| """Get Qiskit IBM URL.""" | ||
| return os.environ.get("QISKIT_IBM_URL", None) | ||
|
|
||
| @classmethod | ||
| def qiskit_ibm_instance(cls) -> Optional[str]: | ||
| """Get Qiskit IBM instance.""" | ||
| return os.environ.get("QISKIT_IBM_INSTANCE", None) | ||
|
|
||
| # Data Configuration | ||
| @classmethod | ||
| def data_path(cls) -> str: | ||
| """Get data path.""" | ||
| return os.environ.get("DATA_PATH", "/data") | ||
|
|
||
| # Request Configuration | ||
| @classmethod | ||
| def requests_timeout(cls) -> int: | ||
| """Get request timeout in seconds.""" | ||
| return int(os.environ.get("REQUESTS_TIMEOUT_OVERRIDE", "30")) | ||
|
|
||
| @classmethod | ||
| def requests_streaming_timeout(cls) -> int: | ||
| """Get streaming request timeout in seconds.""" | ||
| return int(os.environ.get("REQUESTS_TIMEOUT_OVERRIDE", "60")) | ||
|
Contributor
There was a problem hiding this comment. These two config values have the same environment variable |
||
|
|
||
| # Artifact Configuration | ||
| @classmethod | ||
| def max_artifact_file_size_mb(cls) -> int: | ||
| """Get maximum artifact file size in MB.""" | ||
| return int(os.environ.get("MAX_ARTIFACT_FILE_SIZE_MB_OVERRIDE", "50")) | ||
|
|
||
| # IBM Serverless Configuration | ||
| @classmethod | ||
| def ibm_serverless_host_url(cls) -> str: | ||
|
Contributor
There was a problem hiding this comment. I should be ibm_serverless_host_url_override |
||
| """Get IBM Serverless host URL.""" | ||
| return os.environ.get( | ||
| "IBM_SERVERLESS_HOST_URL_OVERRIDE", | ||
| "https://qiskit-serverless.quantum.ibm.com", | ||
| ) | ||
|
|
||
| # Access Configuration | ||
| @classmethod | ||
| def is_trial(cls) -> bool: | ||
| """Check if running in trial mode.""" | ||
| return os.environ.get("ENV_ACCESS_TRIAL") == "True" | ||
|
|
||
| # Test Configuration | ||
| @classmethod | ||
| def in_test(cls) -> Optional[str]: | ||
| """Check if running in test mode.""" | ||
| return os.environ.get("IN_TEST") | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why have two methods (where one calls the other) instead of just one?