22639255938

Committed 03 Mar 2026 07:30PM UTC coverage: 78.369% (-10.6%) from 88.948%

Build # 22639255938

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #9 from CBIIT/data-retriever-service

updates to handle CCDI use case

Coverage Stats

127 of 289 new or added lines in 8 files covered. (43.94%)

1 existing line in 1 file now uncovered.

913 of 1165 relevant lines covered (78.37%)

0.78 hits per line

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

45.63

/core/fetcher.py

import logging
import requests
from typing import Optional, Union

logger = logging.getLogger(__name__)


REQUEST_TIMEOUT = (5, 30)  # (connect_timeout, read_timeout)


def fetch_from_source(source: dict) -> Optional[list]:
    """Routes external data source fetching to appropriate fetching function
    based on source config.

    Args:
        source (dict): Config for external data source.

    Returns:
        Optional[list]: Data fetched from the source, or None if no data was retrieved.
    """
    source_name = source.get("name", "")
    source_type = source.get("type", "").lower()

    logger.info(f"Starting fetch from source: {source_name} (type: {source_type})")

    try:
        if source_type == "rest":
            if "discovery" in source:
                logger.debug(f"Using two-part fetch for source: {source_name}")
                data = do_discovery_then_fetch(source)
            else:
                logger.debug(f"Using direct fetch for source: {source_name}")
                data = fetch_direct(source)
        elif source_type == "graphql":
            logger.debug(f"Using GraphQL fetch for source: {source_name}")
            data = fetch_graphql(source)
        elif source_type == "rest_raw":
            logger.debug(f"Using raw fetch for source: {source_name}")
            data = fetch_raw(source)
        else:
            logger.warning(
                f"Unknown source type '{source_type}' for source: {source_name}"
            )
            return None

        if data is None or data == []:
            logger.warning(f"No data returned from source: {source_name}")
        else:
            logger.info(
                f"Successfully fetched data from source: {source_name} (records: {len(data) if isinstance(data, list) else 'n/a'})"
            )

        # associate source name with fetched data
        if isinstance(data, list):
            for record in data:
                if isinstance(record, dict) and "repository" not in record:
                    record["repository"] = source_name
        elif isinstance(data, dict) and "repository" not in data:
            data["repository"] = source_name

        return data
    except Exception as e:
        logger.error(
            f"Failed to fetch data from source: {source_name}. Error: {e}",
            exc_info=True,
        )
        return None


def fetch_direct(source: dict) -> list:
    """Performs direct fetch (i.e. GET request) using the endpoint specified
    in the source config.

    Args:
        source (dict): Config for external data source.

    Returns:
        list: Data fetched from the source.

    Raises:
        RuntimeError: If the request fails or response is invalid.
    """
    source_name = source.get("name", "<unknown>")

    logger.info(f"Starting direct fetch for source: {source_name}")

    try:
        source_url = f"{source['api_base_url']}{source['endpoint']}"
        logger.debug(f"Request URL: {source_url}")
        response = requests.get(source_url, timeout=REQUEST_TIMEOUT)
        if not response.ok:
            logger.error(
                f"Direct fetch failed for source '{source_name}': {response.status_code} {response.reason}"
            )
            raise RuntimeError(
                f"Fetch failed: {response.status_code} {response.reason}"
            )
        data = extract_response_data(source, response.json())
        if (
            isinstance(data, list)
            and "filter_prefix" in source
            and "match_key" in source
        ):
            filter_prefix = source["filter_prefix"]
            match_key = source["match_key"]
            logger.debug(
                f"Filtering fetched data for prefix '{filter_prefix}' on key '{match_key}'"
            )
            data = [
                item
                for item in data
                if item.get(match_key, "").startswith(filter_prefix)
            ]
        record_count = len(data) if isinstance(data, list) else 1
        logger.info(f"Fetched {record_count} records from source: {source_name}")
        return data
    except requests.exceptions.Timeout as e:
        logger.warning(
            f"Request timed out for source {source['name']} (url={source_url})"
        )
        return []
    except requests.exceptions.RequestException as e:
        logger.error(f"RequestException for source {source_name}: {e}")
        raise RuntimeError(f"Request failed for source {source_name}: {e}") from e
    except ValueError as e:
        logger.error(f"Invalid JSON response for source {source_name}: {e}")
        raise RuntimeError(
            f"Invalid JSON response for source {source_name}: {e}"
        ) from e


def fetch_raw(source: dict) -> list:
    """Fetches data from the given REST endpoint without filtering or entity matching.
    Intended for external sources that should be ingested as-is.

    On request timeout, returns whatever data has been fetched so far (e.g., partial
    results in a paginated scenario).

    Args:
        source (dict): Config for external data source.

    Returns:
        list: Data fetched from the source (may be partial if timeout occurs).

    Raises:
        RuntimeError: If the request fails (non-timeout) or response is invalid.
    """
    source_name = source.get("name", "")
    logger.info(f"Starting raw fetch for source: {source_name}")

    all_data = []
    page = 1
    max_pages = None
    source_url = None

    try:
        source_url = f"{source['api_base_url']}{source['endpoint']}"
        while True:
            logger.debug(f"Request URL: {source_url}")
            response = requests.get(source_url, timeout=REQUEST_TIMEOUT)
            if not response.ok:
                logger.error(
                    f"Raw fetch failed for source '{source_name}': {response.status_code} {response.reason}"
                )
                raise RuntimeError(
                    f"Raw fetch failed: {response.status_code} {response.reason}"
                )
            data = extract_response_data(source, response.json())

            # add current page data
            if isinstance(data, list):
                all_data.extend(data)
            else:
                all_data.append(data)

            # check pagination info
            link_header = response.headers.get("Link", "")
            total_pages_header = response.headers.get("X-Wp-TotalPages", "")

            if total_pages_header and max_pages is None:
                try:
                    max_pages = int(total_pages_header)
                except ValueError:
                    logger.warning(
                        f"Invalid X-Wp-TotalPages header value '{total_pages_header}' for source {source_name}; falling back to Link header parsing for pagination."
                    )

            if max_pages and page >= max_pages:
                break

            next_url = get_next_link(link_header)
            if not next_url:
                break

            if page >= 1000:
                logger.warning(
                    f"Aborting raw fetch for source {source_name} after 1000 pages (possible infinite loop)."
                )
                break

            page += 1
            source_url = next_url

        logger.info(f"Fetched {len(all_data)} records from source: {source_name}")
        return all_data
    except requests.exceptions.Timeout as e:
        logger.warning(
            f"Request timed out for source {source_name} (url={source_url}); "
            f"returning {len(all_data)} records fetched so far."
        )
        return all_data
    except requests.exceptions.RequestException as e:
        logger.error(f"RequestException for source {source_name}: {e}")
        raise RuntimeError(f"Request failed for source {source_name}: {e}") from e
    except ValueError as e:
        logger.error(f"Invalid JSON response for source {source_name}: {e}")
        raise RuntimeError(
            f"Invalid JSON response for source {source_name}: {e}"
        ) from e


def do_discovery_then_fetch(source: dict) -> list:
    """Performs a two-step fetch process, where data from a 'discovery'
    endpoint is used to generate one or more follow-up fetch requests.

    Args:
        source (dict): Config for external data source.

    Returns:
        list: Data fetched from the source.
    """
    discovery = source["discovery"]
    match_key = discovery["match_key"]
    filter_prefix = discovery["filter_prefix"]
    api_base_url = source["api_base_url"]
    discovery_url = f"{api_base_url}{discovery['endpoint']}"

    try:
        logger.debug(f"Starting fetch from discovery URL: {discovery_url}")
        discovery_res = requests.get(discovery_url, timeout=REQUEST_TIMEOUT)
        if not discovery_res.ok:
            logger.error(
                f"Discovery fetch failed for source '{source['name']}': {discovery_res.status_code} {discovery_res.reason}"
            )
            raise RuntimeError(
                f"Fetch failed: {discovery_res.status_code} {discovery_res.reason}"
            )
        discovery_data = extract_response_data(source, discovery_res.json())
        logger.debug(
            f"Filtering fetched data for prefix '{filter_prefix}' on key '{match_key}'"
        )
        filtered_discovery_data = [
            item[match_key]
            for item in discovery_data
            if item[match_key].startswith(filter_prefix)
        ]
        logger.debug("Starting fetch phase...")
        fetch_data = []
        for match in filtered_discovery_data:
            try:
                param = source["fetch"]["key_param"]
                endpoint = source["fetch"]["endpoint_template"].format(**{param: match})
                fetch_url = f"{api_base_url}{endpoint}"
            except KeyError as e:
                logger.error(f"Missing key in 'endpoint_template': {e}")
                raise ValueError(f"Missing key in 'endpoint_template': {e}")
            try:
                res = requests.get(fetch_url, timeout=REQUEST_TIMEOUT)
                if not res.ok:
                    logger.error(
                        f"Fetch phase failed for {fetch_url}: {res.status_code} {res.reason}"
                    )
                    raise RuntimeError(f"Fetch failed: {res.status_code} {res.reason}")
                data = extract_response_data(source, res.json())
                logger.debug(
                    f"Successfully fetched data for match '{match}' from URL: {fetch_url}"
                )
                fetch_data.append(data)
            except requests.exceptions.Timeout as e:
                logger.warning(
                    f"Request timed out for source {source['name']} (url={fetch_url})"
                )
                continue
        logger.info(
            f"Fetched {len(fetch_data)} batches of data from source: {source['name']}"
        )
        return fetch_data
    except requests.exceptions.Timeout as e:
        logger.warning(
            f"Request timed out for source {source['name']} (url={discovery_url})"
        )
        return []
    except requests.exceptions.RequestException as e:
        logger.error(f"RequestException for source {source['name']}: {e}")
        raise RuntimeError(f"Request failed for source {source['name']}: {e}") from e
    except ValueError as e:
        logger.error(f"Invalid JSON response for source {source['name']}: {e}")
        raise RuntimeError(
            f"Invalid JSON response for source {source['name']}: {e}"
        ) from e


def fetch_graphql(source: dict) -> list:
    """Performs data fetch from a GraphQL endpoint via a POST request.

    Args:
        source (dict): Config for external data source.

    Returns:
        list: Data fetched from the GraphQL source.

    Raises:
        RuntimeError: If the GraphQL query fails or returns an error.
    """
    logger.info(f"Starting GraphQL fetch for source: {source['name']}")

    try:
        source_url = f"{source['api_base_url']}{source['endpoint']}"
        logger.debug(f"GraphQL fetch URL: {source_url}")
        response = requests.post(
            url=source_url, json={"query": source["query"]}, timeout=REQUEST_TIMEOUT
        )
        if not response.ok:
            logger.error(
                f"GraphQL fetch failed for {source_url}: {response.status_code} {response.reason}"
            )
            raise RuntimeError(
                f"GraphQL fetch failed: {response.status_code} {response.reason}"
            )
        data = response.json()
        logger.info(f"GraphQL fetch successful for source: {source['name']}")
        return extract_response_data(source, data)
    except requests.exceptions.Timeout as e:
        logger.warning(
            f"Request timed out for source {source['name']} (url={source_url})"
        )
        return []
    except requests.exceptions.RequestException as e:
        logger.error(f"RequestException for source {source['name']}: {e}")
        raise RuntimeError(f"Request failed for source {source['name']}: {e}") from e
    except ValueError as e:
        logger.error(f"Invalid JSON response for source {source['name']}: {e}")
        raise RuntimeError(
            f"Invalid JSON response for source {source['name']}: {e}"
        ) from e


def extract_response_data(source: dict, response_json: dict) -> Union[list, dict]:
    """Extracts relevant data from the full JSON response using the 'response_data_key'
    provided in source config.

    Args:
        source (dict): Config for external data source.
        response_json (dict): JSON object returned from data fetch.

    Returns:
        Union[list, dict]: Portion of response containing relevant data.
    """
    key = source.get("response_data_key")
    if not key:
        return response_json
    for part in key.split("."):
        if not isinstance(response_json, dict):
            return {}
        response_json = response_json.get(part, {})
    return response_json


def get_next_link(link_header: str) -> Optional[str]:
    """Parses the 'Link' HTTP header to extract the URL for the 'next' page.

    Args:
        link_header (str): The 'Link' HTTP header string.
    Returns:
        Optional[str]: The URL for the next page, or None if not found.
    """
    if not link_header:
        return None

    entries = [e.strip() for e in link_header.split(",")]
    for entry in entries:
        # expect format: <url>; rel="next"
        if ";" not in entry:
            continue
        url_part, *params = [p.strip() for p in entry.split(";")]
        if not (url_part.startswith("<") and url_part.endswith(">")):
            continue

        url = url_part[1:-1]
        for p in params:
            if p.startswith("rel="):
                rel_value = p[4:].strip('"')
                if rel_value == "next":
                    return url

    return None

1	import logging	1✔
2	import requests	1✔
3	from typing import Optional, Union	1✔
4
5	logger = logging.getLogger(__name__)	1✔
6
7
8	REQUEST_TIMEOUT = (5, 30) # (connect_timeout, read_timeout)	1✔
9
10
11	def fetch_from_source(source: dict) -> Optional[list]:	1✔
12	"""Routes external data source fetching to appropriate fetching function
13	based on source config.
14
15	Args:
16	source (dict): Config for external data source.
17
18	Returns:
19	Optional[list]: Data fetched from the source, or None if no data was retrieved.
20	"""
21	source_name = source.get("name", "")	1✔
22	source_type = source.get("type", "").lower()	1✔
23
24	logger.info(f"Starting fetch from source: {source_name} (type: {source_type})")	1✔
25
26	try:	1✔
27	if source_type == "rest":	1✔
28	if "discovery" in source:	1✔
29	logger.debug(f"Using two-part fetch for source: {source_name}")	1✔
30	data = do_discovery_then_fetch(source)	1✔
31	else:
32	logger.debug(f"Using direct fetch for source: {source_name}")	1✔
33	data = fetch_direct(source)	1✔
34	elif source_type == "graphql":	1✔
35	logger.debug(f"Using GraphQL fetch for source: {source_name}")	1✔
36	data = fetch_graphql(source)	1✔
37	elif source_type == "rest_raw":	1✔
NEW 38	logger.debug(f"Using raw fetch for source: {source_name}")	×
NEW 39	data = fetch_raw(source)	×
40	else:
41	logger.warning(	1✔
42	f"Unknown source type '{source_type}' for source: {source_name}"
43	)
44	return None	1✔
45
46	if data is None or data == []:	1✔
47	logger.warning(f"No data returned from source: {source_name}")	×
48	else:
49	logger.info(	1✔
50	f"Successfully fetched data from source: {source_name} (records: {len(data) if isinstance(data, list) else 'n/a'})"
51	)
52
53	# associate source name with fetched data
54	if isinstance(data, list):	1✔
55	for record in data:	1✔
56	if isinstance(record, dict) and "repository" not in record:	1✔
57	record["repository"] = source_name	1✔
NEW 58	elif isinstance(data, dict) and "repository" not in data:	×
NEW 59	data["repository"] = source_name	×
60
61	return data	1✔
62	except Exception as e:	×
63	logger.error(	×
64	f"Failed to fetch data from source: {source_name}. Error: {e}",
65	exc_info=True,
66	)
67	return None	×
68
69
70	def fetch_direct(source: dict) -> list:	1✔
71	"""Performs direct fetch (i.e. GET request) using the endpoint specified
72	in the source config.
73
74	Args:
75	source (dict): Config for external data source.
76
77	Returns:
78	list: Data fetched from the source.
79
80	Raises:
81	RuntimeError: If the request fails or response is invalid.
82	"""
83	source_name = source.get("name", "<unknown>")	1✔
84
85	logger.info(f"Starting direct fetch for source: {source_name}")	1✔
86
87	try:	1✔
88	source_url = f"{source['api_base_url']}{source['endpoint']}"	1✔
89	logger.debug(f"Request URL: {source_url}")	1✔
90	response = requests.get(source_url, timeout=REQUEST_TIMEOUT)	1✔
91	if not response.ok:	1✔
92	logger.error(	×
93	f"Direct fetch failed for source '{source_name}': {response.status_code} {response.reason}"
94	)
95	raise RuntimeError(	×
96	f"Fetch failed: {response.status_code} {response.reason}"
97	)
98	data = extract_response_data(source, response.json())	1✔
99	if (	1✔
100	isinstance(data, list)
101	and "filter_prefix" in source
102	and "match_key" in source
103	):
104	filter_prefix = source["filter_prefix"]	1✔
105	match_key = source["match_key"]	1✔
106	logger.debug(	1✔
107	f"Filtering fetched data for prefix '{filter_prefix}' on key '{match_key}'"
108	)
109	data = [	1✔
110	item
111	for item in data
112	if item.get(match_key, "").startswith(filter_prefix)
113	]
114	record_count = len(data) if isinstance(data, list) else 1	1✔
115	logger.info(f"Fetched {record_count} records from source: {source_name}")	1✔
116	return data	1✔
NEW 117	except requests.exceptions.Timeout as e:	×
NEW 118	logger.warning(	×
119	f"Request timed out for source {source['name']} (url={source_url})"
120	)
NEW 121	return []	×
122	except requests.exceptions.RequestException as e:	×
123	logger.error(f"RequestException for source {source_name}: {e}")	×
NEW 124	raise RuntimeError(f"Request failed for source {source_name}: {e}") from e	×
NEW 125	except ValueError as e:	×
NEW 126	logger.error(f"Invalid JSON response for source {source_name}: {e}")	×
NEW 127	raise RuntimeError(	×
128	f"Invalid JSON response for source {source_name}: {e}"
129	) from e
130
131
132	def fetch_raw(source: dict) -> list:	1✔
133	"""Fetches data from the given REST endpoint without filtering or entity matching.
134	Intended for external sources that should be ingested as-is.
135
136	On request timeout, returns whatever data has been fetched so far (e.g., partial
137	results in a paginated scenario).
138
139	Args:
140	source (dict): Config for external data source.
141
142	Returns:
143	list: Data fetched from the source (may be partial if timeout occurs).
144
145	Raises:
146	RuntimeError: If the request fails (non-timeout) or response is invalid.
147	"""
NEW 148	source_name = source.get("name", "")	×
NEW 149	logger.info(f"Starting raw fetch for source: {source_name}")	×
150
NEW 151	all_data = []	×
NEW 152	page = 1	×
NEW 153	max_pages = None	×
NEW 154	source_url = None	×
155
NEW 156	try:	×
NEW 157	source_url = f"{source['api_base_url']}{source['endpoint']}"	×
NEW 158	while True:	×
NEW 159	logger.debug(f"Request URL: {source_url}")	×
NEW 160	response = requests.get(source_url, timeout=REQUEST_TIMEOUT)	×
NEW 161	if not response.ok:	×
NEW 162	logger.error(	×
163	f"Raw fetch failed for source '{source_name}': {response.status_code} {response.reason}"
164	)
NEW 165	raise RuntimeError(	×
166	f"Raw fetch failed: {response.status_code} {response.reason}"
167	)
NEW 168	data = extract_response_data(source, response.json())	×
169
170	# add current page data
NEW 171	if isinstance(data, list):	×
NEW 172	all_data.extend(data)	×
173	else:
NEW 174	all_data.append(data)	×
175
176	# check pagination info
NEW 177	link_header = response.headers.get("Link", "")	×
NEW 178	total_pages_header = response.headers.get("X-Wp-TotalPages", "")	×
179
NEW 180	if total_pages_header and max_pages is None:	×
NEW 181	try:	×
NEW 182	max_pages = int(total_pages_header)	×
NEW 183	except ValueError:	×
NEW 184	logger.warning(	×
185	f"Invalid X-Wp-TotalPages header value '{total_pages_header}' for source {source_name}; falling back to Link header parsing for pagination."
186	)
187
NEW 188	if max_pages and page >= max_pages:	×
NEW 189	break	×
190
NEW 191	next_url = get_next_link(link_header)	×
NEW 192	if not next_url:	×
NEW 193	break	×
194
NEW 195	if page >= 1000:	×
NEW 196	logger.warning(	×
197	f"Aborting raw fetch for source {source_name} after 1000 pages (possible infinite loop)."
198	)
NEW 199	break	×
200
NEW 201	page += 1	×
NEW 202	source_url = next_url	×
203
NEW 204	logger.info(f"Fetched {len(all_data)} records from source: {source_name}")	×
NEW 205	return all_data	×
206	except requests.exceptions.Timeout as e:	×
207	logger.warning(	×
208	f"Request timed out for source {source_name} (url={source_url}); "
209	f"returning {len(all_data)} records fetched so far."
210	)
NEW 211	return all_data	×
NEW 212	except requests.exceptions.RequestException as e:	×
NEW 213	logger.error(f"RequestException for source {source_name}: {e}")	×
NEW 214	raise RuntimeError(f"Request failed for source {source_name}: {e}") from e	×
215	except ValueError as e:	×
216	logger.error(f"Invalid JSON response for source {source_name}: {e}")	×
NEW 217	raise RuntimeError(	×
218	f"Invalid JSON response for source {source_name}: {e}"
219	) from e
220
221
222	def do_discovery_then_fetch(source: dict) -> list:	1✔
223	"""Performs a two-step fetch process, where data from a 'discovery'
224	endpoint is used to generate one or more follow-up fetch requests.
225
226	Args:
227	source (dict): Config for external data source.
228
229	Returns:
230	list: Data fetched from the source.
231	"""
232	discovery = source["discovery"]	1✔
233	match_key = discovery["match_key"]	1✔
234	filter_prefix = discovery["filter_prefix"]	1✔
235	api_base_url = source["api_base_url"]	1✔
236	discovery_url = f"{api_base_url}{discovery['endpoint']}"	1✔
237
238	try:	1✔
239	logger.debug(f"Starting fetch from discovery URL: {discovery_url}")	1✔
240	discovery_res = requests.get(discovery_url, timeout=REQUEST_TIMEOUT)	1✔
241	if not discovery_res.ok:	1✔
242	logger.error(	×
243	f"Discovery fetch failed for source '{source['name']}': {discovery_res.status_code} {discovery_res.reason}"
244	)
245	raise RuntimeError(	×
246	f"Fetch failed: {discovery_res.status_code} {discovery_res.reason}"
247	)
248	discovery_data = extract_response_data(source, discovery_res.json())	1✔
249	logger.debug(	1✔
250	f"Filtering fetched data for prefix '{filter_prefix}' on key '{match_key}'"
251	)
252	filtered_discovery_data = [	1✔
253	item[match_key]
254	for item in discovery_data
255	if item[match_key].startswith(filter_prefix)
256	]
257	logger.debug("Starting fetch phase...")	1✔
258	fetch_data = []	1✔
259	for match in filtered_discovery_data:	1✔
260	try:	1✔
261	param = source["fetch"]["key_param"]	1✔
262	endpoint = source["fetch"]["endpoint_template"].format(**{param: match})	1✔
263	fetch_url = f"{api_base_url}{endpoint}"	1✔
264	except KeyError as e:	×
265	logger.error(f"Missing key in 'endpoint_template': {e}")	×
266	raise ValueError(f"Missing key in 'endpoint_template': {e}")	×
267	try:	1✔
268	res = requests.get(fetch_url, timeout=REQUEST_TIMEOUT)	1✔
269	if not res.ok:	1✔
270	logger.error(	×
271	f"Fetch phase failed for {fetch_url}: {res.status_code} {res.reason}"
272	)
273	raise RuntimeError(f"Fetch failed: {res.status_code} {res.reason}")	×
274	data = extract_response_data(source, res.json())	1✔
275	logger.debug(	1✔
276	f"Successfully fetched data for match '{match}' from URL: {fetch_url}"
277	)
278	fetch_data.append(data)	1✔
279	except requests.exceptions.Timeout as e:	×
280	logger.warning(	×
281	f"Request timed out for source {source['name']} (url={fetch_url})"
282	)
283	continue	×
284	logger.info(	1✔
285	f"Fetched {len(fetch_data)} batches of data from source: {source['name']}"
286	)
287	return fetch_data	1✔
288	except requests.exceptions.Timeout as e:	×
289	logger.warning(	×
290	f"Request timed out for source {source['name']} (url={discovery_url})"
291	)
NEW 292	return []	×
NEW 293	except requests.exceptions.RequestException as e:	×
NEW 294	logger.error(f"RequestException for source {source['name']}: {e}")	×
NEW 295	raise RuntimeError(f"Request failed for source {source['name']}: {e}") from e	×
296	except ValueError as e:	×
297	logger.error(f"Invalid JSON response for source {source['name']}: {e}")	×
NEW 298	raise RuntimeError(	×
299	f"Invalid JSON response for source {source['name']}: {e}"
300	) from e
301
302
303	def fetch_graphql(source: dict) -> list:	1✔
304	"""Performs data fetch from a GraphQL endpoint via a POST request.
305
306	Args:
307	source (dict): Config for external data source.
308
309	Returns:
310	list: Data fetched from the GraphQL source.
311
312	Raises:
313	RuntimeError: If the GraphQL query fails or returns an error.
314	"""
315	logger.info(f"Starting GraphQL fetch for source: {source['name']}")	1✔
316
317	try:	1✔
318	source_url = f"{source['api_base_url']}{source['endpoint']}"	1✔
319	logger.debug(f"GraphQL fetch URL: {source_url}")	1✔
320	response = requests.post(	1✔
321	url=source_url, json={"query": source["query"]}, timeout=REQUEST_TIMEOUT
322	)
323	if not response.ok:	1✔
324	logger.error(	×
325	f"GraphQL fetch failed for {source_url}: {response.status_code} {response.reason}"
326	)
327	raise RuntimeError(	×
328	f"GraphQL fetch failed: {response.status_code} {response.reason}"
329	)
330	data = response.json()	1✔
331	logger.info(f"GraphQL fetch successful for source: {source['name']}")	1✔
332	return extract_response_data(source, data)	1✔
333	except requests.exceptions.Timeout as e:	×
334	logger.warning(	×
335	f"Request timed out for source {source['name']} (url={source_url})"
336	)
NEW 337	return []	×
NEW 338	except requests.exceptions.RequestException as e:	×
NEW 339	logger.error(f"RequestException for source {source['name']}: {e}")	×
NEW 340	raise RuntimeError(f"Request failed for source {source['name']}: {e}") from e	×
341	except ValueError as e:	×
342	logger.error(f"Invalid JSON response for source {source['name']}: {e}")	×
NEW 343	raise RuntimeError(	×
344	f"Invalid JSON response for source {source['name']}: {e}"
345	) from e
346
347
348	def extract_response_data(source: dict, response_json: dict) -> Union[list, dict]:	1✔
349	"""Extracts relevant data from the full JSON response using the 'response_data_key'
350	provided in source config.
351
352	Args:
353	source (dict): Config for external data source.
354	response_json (dict): JSON object returned from data fetch.
355
356	Returns:
357	Union[list, dict]: Portion of response containing relevant data.
358	"""
359	key = source.get("response_data_key")	1✔
360	if not key:	1✔
361	return response_json	1✔
362	for part in key.split("."):	1✔
363	if not isinstance(response_json, dict):	1✔
364	return {}	×
365	response_json = response_json.get(part, {})	1✔
366	return response_json	1✔
367
368
369	def get_next_link(link_header: str) -> Optional[str]:	1✔
370	"""Parses the 'Link' HTTP header to extract the URL for the 'next' page.
371
372	Args:
373	link_header (str): The 'Link' HTTP header string.
374	Returns:
375	Optional[str]: The URL for the next page, or None if not found.
376	"""
NEW 377	if not link_header:	×
NEW 378	return None	×
379
NEW 380	entries = [e.strip() for e in link_header.split(",")]	×
NEW 381	for entry in entries:	×
382	# expect format: <url>; rel="next"
NEW 383	if ";" not in entry:	×
NEW 384	continue	×
NEW 385	url_part, *params = [p.strip() for p in entry.split(";")]	×
NEW 386	if not (url_part.startswith("<") and url_part.endswith(">")):	×
NEW 387	continue	×
388
NEW 389	url = url_part[1:-1]	×
NEW 390	for p in params:	×
NEW 391	if p.startswith("rel="):	×
NEW 392	rel_value = p[4:].strip('"')	×
NEW 393	if rel_value == "next":	×
NEW 394	return url	×
395
NEW 396	return None	×

CBIIT / crdc-icdc-data-retriever / 22639255938

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