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.
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import anyio
|
|
from licaeros import LicensedSession, device_fingerprint_hex
|
|
from src.config import AEROS_BASE_URL, AEROS_LICENSE_ID, AEROS_LICENSE_SECRET
|
|
import logging
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Initialize a global session if possible, or create on demand
|
|
_aeros_session = None
|
|
|
|
def get_aeros_session():
|
|
global _aeros_session
|
|
if _aeros_session is None:
|
|
log.info(f"Initializing LicensedSession with base URL: {AEROS_BASE_URL}")
|
|
log.info(f"Encrypted Device ID: {device_fingerprint_hex()}")
|
|
_aeros_session = LicensedSession(
|
|
api_base=AEROS_BASE_URL,
|
|
license_id=AEROS_LICENSE_ID,
|
|
license_secret=AEROS_LICENSE_SECRET,
|
|
)
|
|
return _aeros_session
|
|
|
|
async def aeros_post(path: str, json: dict = None, **kwargs):
|
|
"""
|
|
Asynchronous wrapper for LicensedSession.post
|
|
"""
|
|
session = get_aeros_session()
|
|
# LicensedSession might not be async-compatible, so we run it in a thread
|
|
response = await anyio.to_thread.run_sync(
|
|
lambda: session.post(path, json)
|
|
)
|
|
return response
|
|
|
|
|
|
async def aeros_file_upload(path, file, field_name, filename):
|
|
session = get_aeros_session()
|
|
response = await anyio.to_thread.run_sync(
|
|
lambda: session.post_multipart(path, file, field_name, filename)
|
|
)
|
|
return response |