You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
122 lines
4.7 KiB
Python
122 lines
4.7 KiB
Python
import anyio
|
|
import logging
|
|
import httpx
|
|
from src.config import (
|
|
AEROS_BASE_URL,
|
|
WINDOWS_AEROS_BASE_URL,
|
|
VAULT_URL,
|
|
ROLE_ID,
|
|
SECRET_ID,
|
|
AEROS_SECRET_PATH,
|
|
AEROS_LICENSE_ID,
|
|
AEROS_LICENSE_SECRET,
|
|
USE_LICENSE_APP
|
|
)
|
|
from src.utils import get_vault_secrets
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Try to import licaeros gracefully
|
|
try:
|
|
from licaeros import LicensedSession, device_fingerprint_hex
|
|
HAS_LICAEROS = True
|
|
except ImportError:
|
|
HAS_LICAEROS = False
|
|
log.warning("licaeros library not found. Falling back to direct httpx calls if enabled.")
|
|
|
|
# Initialize a global session if possible, or create on demand
|
|
_aeros_session = None
|
|
|
|
def get_aeros_session(base_url):
|
|
global _aeros_session
|
|
|
|
if not USE_LICENSE_APP or not HAS_LICAEROS:
|
|
if _aeros_session is None or not isinstance(_aeros_session, httpx.Client) or str(_aeros_session.base_url) != str(base_url):
|
|
log.info(f"Using direct httpx.Client for base URL: {base_url}")
|
|
_aeros_session = httpx.Client(base_url=base_url, timeout=300.0)
|
|
return _aeros_session
|
|
|
|
# License App path - only proceed if USE_LICENSE_APP is True
|
|
license_id = AEROS_LICENSE_ID
|
|
license_secret = AEROS_LICENSE_SECRET
|
|
|
|
# If vault is configured, try to get from there
|
|
if VAULT_URL and ROLE_ID and SECRET_ID and AEROS_SECRET_PATH:
|
|
results = get_vault_secrets(
|
|
vault_url=VAULT_URL,
|
|
role_id=ROLE_ID,
|
|
secret_id=SECRET_ID,
|
|
secret_path=AEROS_SECRET_PATH,
|
|
secret_keys_to_be_returned=['aeros_license_id', 'aeros_license_secret']
|
|
)
|
|
if results:
|
|
license_id = results['aeros_license_id']
|
|
license_secret = results['aeros_license_secret']
|
|
log.info("Aeros license retrieved from Vault")
|
|
else:
|
|
log.warning("Failed to get Aeros license from Vault, trying local env fallback")
|
|
|
|
if not license_id or not license_secret:
|
|
log.warning("Aeros license ID or Secret not provided. Falling back to direct httpx.")
|
|
if _aeros_session is None or not isinstance(_aeros_session, httpx.Client):
|
|
_aeros_session = httpx.Client(base_url=base_url, timeout=300.0)
|
|
return _aeros_session
|
|
|
|
if _aeros_session is None or isinstance(_aeros_session, httpx.Client):
|
|
log.info(f"Initializing LicensedSession with base URL: {base_url}")
|
|
log.info(f"Encrypted Device ID: {device_fingerprint_hex()}")
|
|
_aeros_session = LicensedSession(
|
|
api_base=base_url,
|
|
license_id=license_id,
|
|
license_secret=license_secret,
|
|
timeout=1000
|
|
)
|
|
return _aeros_session
|
|
|
|
async def aeros_post(path: str, json=None, data=None, **kwargs):
|
|
"""
|
|
Asynchronous wrapper for Aeros requests (Licensed or Direct)
|
|
"""
|
|
# If not using license app, fetch from AEROS_BASE_URL direcly
|
|
target_base_url = WINDOWS_AEROS_BASE_URL if USE_LICENSE_APP and HAS_LICAEROS else AEROS_BASE_URL
|
|
session = get_aeros_session(target_base_url)
|
|
|
|
# Path logic: License app often uses /api/aeros prefix, direct might not.
|
|
# However, to maintain compatibility with existing code calling this with path="/api/Project/ImportAROFile"
|
|
# we should decide if we need to adjust the path.
|
|
# User said: "system still fetch to aeros direcly not to licence app then lincense app forward it to aeros"
|
|
# This implies the license app acts as a proxy for /api/aeros/ paths.
|
|
|
|
if USE_LICENSE_APP and HAS_LICAEROS:
|
|
url = f"/api/aeros{path}"
|
|
response = await anyio.to_thread.run_sync(
|
|
lambda: session.post(url, json_data=json, data=data, headers=kwargs.get("headers"))
|
|
)
|
|
else:
|
|
# Direct fetch from Aeros might have different path structure?
|
|
# Assuming direct Aeros path is same as what the license app proxies to.
|
|
url = path
|
|
response = await anyio.to_thread.run_sync(
|
|
lambda: session.post(url, json=json, data=data, headers=kwargs.get("headers"))
|
|
)
|
|
|
|
return response
|
|
|
|
|
|
async def aeros_file_upload(path, file, field_name, filename, base_url=None):
|
|
target_base_url = WINDOWS_AEROS_BASE_URL if USE_LICENSE_APP and HAS_LICAEROS else (base_url or AEROS_BASE_URL)
|
|
session = get_aeros_session(target_base_url)
|
|
|
|
if USE_LICENSE_APP and HAS_LICAEROS:
|
|
url = f"/api/aeros{path}"
|
|
response = await anyio.to_thread.run_sync(
|
|
lambda: session.post_multipart(url, file, field_name, filename)
|
|
)
|
|
else:
|
|
url = path
|
|
files = {field_name: (filename, file)}
|
|
response = await anyio.to_thread.run_sync(
|
|
lambda: session.post(url, files=files)
|
|
)
|
|
|
|
return response |