"""
Module contains internal lambda http client for usage in lambda
"""
from collections.abc import Awaitable
import aiohttp
from cow.utils.lambdas_functions import getK8SLambdaServiceName
from cow.utils.log import logger
from luna3.common.luna_response import LunaResponse
from luna3.common.requests import RequestPayload, makeRequest
[docs]
class Lambdas:
"""Internal lambdas api client"""
[docs]
def __init__(self, accountId: str):
self._accountId = accountId
[docs]
def lambdasRequest(
self,
lambdaId: str | None = None,
lambdaName: str | None = None,
method: str = "GET",
path: str = "main",
namespace: str | None = None,
body: str | bytes | dict | list | aiohttp.Payload | RequestPayload = None,
headers: dict | None = None,
queryParams: dict | None = None,
raiseError: bool | None = False,
**kwargs,
) -> Awaitable[LunaResponse] | LunaResponse:
"""
Make request to other lambda
Args:
lambdaId: lambda id
lambdaName: lambda name
method: HTTP method
path: any additional path parameter will be use as path parameter to lambda
namespace: lambda namespace
body: request body
headers: request headers
queryParams: request query params
raiseError: if request fails, LunaApiException is raised
Returns:
class:`~.LunaResponse` or *asyncio coroutine* with *LunaResponse*.
In body of :class: `~.LunaResponse` json with events will be returned.
"""
if lambdaId is None and lambdaName is None:
raise ValueError("Invalid arguments: one of 'lambdaId' or 'lambdaName' is required")
if lambdaId is not None and lambdaName is not None:
logger.warning("lambdaId and lambdaName were received, only lambdaName will be used.")
if lambdaName is not None:
serviceName = getK8SLambdaServiceName(lambdaName)
else:
serviceName = f"lambda-{lambdaId}"
if namespace is None:
url = f"http://{serviceName}:8000/{path}"
else:
url = f"http://{serviceName}.{namespace}:8000/{path}"
response = makeRequest(
url=url,
method=method.upper(),
headers=headers,
queryParams=queryParams,
body=body,
raiseError=raiseError,
**kwargs,
)
return response