utilizing lambda function url class and new lambda function url invoc…

…ation

src/aibs_informatics_aws_utils/lambda_.py

@@ -1,9 +1,15 @@
-from typing import TYPE_CHECKING, List, Optional, Union
+import json
+from typing import TYPE_CHECKING, List, Literal, Optional, Union
+from urllib.parse import parse_qs
 
+import requests
 from aibs_informatics_core.models.aws.core import AWSAccountId, AWSRegion
-from aibs_informatics_core.models.aws.lambda_ import LambdaFunctionName
+from aibs_informatics_core.models.aws.lambda_ import LambdaFunctionName, LambdaFunctionUrl
+from aibs_informatics_core.models.base import ModelProtocol
 from botocore.exceptions import ClientError
+from requests.auth import AuthBase
 
+from aibs_informatics_aws_utils.auth import IamAWSRequestsAuth
 from aibs_informatics_aws_utils.core import AWSService, get_client_error_code
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -16,8 +22,8 @@
 
 
 def get_lambda_function_url(
-    function_name: Union[LambdaFunctionName, str], region: AWSRegion = None
-) -> Optional[str]:
+    function_name: Union[LambdaFunctionName, str], region: Optional[AWSRegion] = None
+) -> Optional[LambdaFunctionUrl]:
     function_name = LambdaFunctionName(function_name)
 
     lambda_client = get_lambda_client(region=region)
@@ -29,11 +35,11 @@ def get_lambda_function_url(
             return None
         else:
             raise e
-    return response["FunctionUrl"]
+    return LambdaFunctionUrl(response["FunctionUrl"])
 
 
 def get_lambda_function_file_systems(
-    function_name: Union[LambdaFunctionName, str], region: AWSRegion = None
+    function_name: Union[LambdaFunctionName, str], region: Optional[AWSRegion] = None
 ) -> List[FileSystemConfigTypeDef]:
     function_name = LambdaFunctionName(function_name)
 
@@ -44,3 +50,59 @@ def get_lambda_function_file_systems(
     fs_configs = response.get("FileSystemConfigs")
 
     return fs_configs
+
+
+def call_lambda_function_url(
+    function_name: Union[LambdaFunctionName, LambdaFunctionUrl, str],
+    payload: Optional[Union[ModelProtocol, dict, str, bytes]] = None,
+    region: Optional[AWSRegion] = None,
+    headers: Optional[dict] = None,
+    auth: Optional[AuthBase] = None,
+    **request_kwargs,
+) -> dict | str | None:
+    if LambdaFunctionName.is_valid(function_name):
+        function_url = get_lambda_function_url(LambdaFunctionName(function_name), region=region)
+        if function_url is None:
+            raise ValueError(f"Function {function_name} not found")
+        function_url = LambdaFunctionUrl(function_url)
+    elif LambdaFunctionUrl.is_valid(function_name):
+        function_url = LambdaFunctionUrl(function_name)
+    else:
+        raise ValueError(f"Invalid function name or url: {function_name}")
+
+    json_payload: Optional[str] = None
+    if isinstance(payload, (dict, list)):
+        json_payload = json.dumps(payload)
+    elif isinstance(payload, ModelProtocol):
+        json_payload = json.dumps(payload.to_dict())
+    elif isinstance(payload, str):
+        json_payload = payload
+    elif isinstance(payload, bytes):
+        json_payload = payload.decode("utf-8")
+    elif payload is None:
+        pass
+    else:
+        raise ValueError(f"Invalid payload type: {type(payload)}")
+
+    if headers is None:
+        headers = {}
+
+    if auth is None:
+        auth = IamAWSRequestsAuth(service_name="lambda")
+
+    response = requests.request(
+        method="POST" if json_payload else "GET",
+        url=function_url.base_url + function_url.path,
+        json=json_payload,
+        params=function_url.query,
+        headers=headers,
+        auth=auth,
+        **request_kwargs,
+    )
+    if response.ok:
+        if response.headers.get("Content-Type") == "application/json":
+            return response.json()
+        else:
+            return response.text
+    else:
+        response.raise_for_status()