14052210327

Committed 25 Mar 2025 05:15AM UTC coverage: 98.083%. Remained the same

Build # 14052210327

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #165 from SAP/ICNMLSALESSERVICESWDF2-4541-Updating-Changelog-and-version

Updating-Changelog-and-version

Run Details

90 of 95 branches covered (94.74%)

Branch coverage included in aggregate %.

831 of 844 relevant lines covered (98.46%)

4.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.57

/sap/aibus/dar/client/aiapi/dar_ai_api_file_upload_client.py

"""Client for DAR File Upload AI API."""

import urllib.parse
from typing import Any, Callable

import requests
from requests import Response


class DARAIAPIFileUploadClient:
    """Client for DAR File Upload AI API.

    This client provides methods to upload and delete files on the DAR service,
    handling authentication and request preparation internally.
    """

    def __init__(self, base_url: str, get_token: Callable[[], str]):
        """Initialize the DARFileUploadAIAPIClient.

        :param base_url: The base URL of the DAR AI API.
        :param get_token: A callable to fetch the authorization token.
        """
        self.base_url = base_url + "/files"
        self.get_token = get_token

    def delete_file(self, remote_path: str) -> Response:
        """Delete file under defined remote path.

        :param remote_path: The path to the file to delete on the server.

        :returns: The HTTP response object from the API call.
        """
        url = self.base_url + remote_path

        return self._send("DELETE", url)

    def put_file(
        self, local_path: str, remote_path: str, overwrite: bool = False
    ) -> Response:
        """Upload file to the defined remote path.

        :param local_path: The local path of the file to upload.
        :param remote_path: The destination path for the file on the server.
        :param overwrite: Whether to overwrite the file if it already exists,
                              defaults to False.

        :returns: The HTTP response object from the API call.
        """
        url = self.base_url + remote_path
        headers = {
            # Content-Type MUST be application/octet-stream, even when uploading
            # e.g. CSV files
            "Content-Type": "application/octet-stream",
        }
        params = {"overwrite": overwrite}
        with open(local_path, "rb") as file:
            return self._send(
                method="PUT", url=url, headers=headers, params=params, data=file
            )

    def get_file_from_url(self, url: str) -> Response:
        """Download file under defined url.

        :param url: The url to the file that needs to be downloaded

        :returns: The downloaded file on the server from the url.
        """
        return self._send("GET", url)

    def _send(  # pylint: disable=too-many-arguments
        self,
        method: str,
        url: str,
        headers: dict = None,
        data: Any = None,
        params: dict = None,
    ) -> Response:
        """Send an HTTP request using the `requests` library.

        :param method: The HTTP method (e.g., 'GET', 'POST', 'PUT', 'DELETE').
        :param url: The full URL for the request.
        :param headers: The headers for the request,defaults to None.
        :param data: The data payload for the request (e.g., file data),
                     defaults to None.
        :param params: The query parameters for the URL,defaults to None.

        :returns: The HTTP response object from the API call.
        """
        session = requests.Session()
        auth_headers = {
            "Authorization": self.get_token(),
        }
        if headers:
            auth_headers.update(headers)

        req = requests.Request(
            method=method, url=url, headers=auth_headers, data=data, params=params
        )
        prep = req.prepare()
        prep.url = url
        if params:
            prep.url += "?" + urllib.parse.urlencode(params)
        response = session.send(prep, verify=True)
        return response

1	"""Client for DAR File Upload AI API."""
2
3	import urllib.parse	5✔
4	from typing import Any, Callable	5✔
5
6	import requests	5✔
7	from requests import Response	5✔
8
9
10	class DARAIAPIFileUploadClient:	5✔
11	"""Client for DAR File Upload AI API.
12
13	This client provides methods to upload and delete files on the DAR service,
14	handling authentication and request preparation internally.
15	"""
16
17	def __init__(self, base_url: str, get_token: Callable[[], str]):	5✔
18	"""Initialize the DARFileUploadAIAPIClient.
19
20	:param base_url: The base URL of the DAR AI API.
21	:param get_token: A callable to fetch the authorization token.
22	"""
23	self.base_url = base_url + "/files"	5✔
24	self.get_token = get_token	5✔
25
26	def delete_file(self, remote_path: str) -> Response:	5✔
27	"""Delete file under defined remote path.
28
29	:param remote_path: The path to the file to delete on the server.
30
31	:returns: The HTTP response object from the API call.
32	"""
33	url = self.base_url + remote_path	5✔
34
35	return self._send("DELETE", url)	5✔
36
37	def put_file(	5✔
38	self, local_path: str, remote_path: str, overwrite: bool = False
39	) -> Response:
40	"""Upload file to the defined remote path.
41
42	:param local_path: The local path of the file to upload.
43	:param remote_path: The destination path for the file on the server.
44	:param overwrite: Whether to overwrite the file if it already exists,
45	defaults to False.
46
47	:returns: The HTTP response object from the API call.
48	"""
49	url = self.base_url + remote_path	5✔
50	headers = {	5✔
51	# Content-Type MUST be application/octet-stream, even when uploading
52	# e.g. CSV files
53	"Content-Type": "application/octet-stream",
54	}
55	params = {"overwrite": overwrite}	5✔
56	with open(local_path, "rb") as file:	5✔
57	return self._send(	5✔
58	method="PUT", url=url, headers=headers, params=params, data=file
59	)
60
61	def get_file_from_url(self, url: str) -> Response:	5✔
62	"""Download file under defined url.
63
64	:param url: The url to the file that needs to be downloaded
65
66	:returns: The downloaded file on the server from the url.
67	"""
68	return self._send("GET", url)	5✔
69
70	def _send( # pylint: disable=too-many-arguments	5✔
71	self,
72	method: str,
73	url: str,
74	headers: dict = None,
75	data: Any = None,
76	params: dict = None,
77	) -> Response:
78	"""Send an HTTP request using the `requests` library.
79
80	:param method: The HTTP method (e.g., 'GET', 'POST', 'PUT', 'DELETE').
81	:param url: The full URL for the request.
82	:param headers: The headers for the request,defaults to None.
83	:param data: The data payload for the request (e.g., file data),
84	defaults to None.
85	:param params: The query parameters for the URL,defaults to None.
86
87	:returns: The HTTP response object from the API call.
88	"""
89	session = requests.Session()	5✔
90	auth_headers = {	5✔
91	"Authorization": self.get_token(),
92	}
93	if headers:	5!
94	auth_headers.update(headers)	×
95
96	req = requests.Request(	5✔
97	method=method, url=url, headers=auth_headers, data=data, params=params
98	)
99	prep = req.prepare()	5✔
100	prep.url = url	5✔
101	if params:	5!
102	prep.url += "?" + urllib.parse.urlencode(params)	×
103	response = session.send(prep, verify=True)	5✔
104	return response	5✔

SAP / data-attribute-recommendation-python-sdk / 14052210327

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous