Setup a user agent for the download_scorefiles utils (REST API calls to the PGS Catalog)

pgscatalog_utils/download/download_scorefile.py

@@ -33,21 +33,25 @@ def download_scorefile() -> None:
 
     pgs_lst: list[list[str]] = []
 
+    pgsc_calc_info = None
+    if args.pgsc_calc:
+        pgsc_calc_info = args.pgsc_calc
+
     if args.efo:
         logger.debug("--trait set, querying traits")
-        pgs_lst = pgs_lst + [query_trait(x) for x in args.efo]
+        pgs_lst = pgs_lst + [query_trait(x, pgsc_calc_info) for x in args.efo]
 
     if args.pgp:
         logger.debug("--pgp set, querying publications")
-        pgs_lst = pgs_lst + [query_publication(x) for x in args.pgp]
+        pgs_lst = pgs_lst + [query_publication(x, pgsc_calc_info) for x in args.pgp]
 
     if args.pgs:
         logger.debug("--id set, querying scores")
         pgs_lst.append(args.pgs)  # pgs_lst: a list containing up to three flat lists
 
     pgs_id: list[str] = list(set(reduce(lambda x, y: x + y, pgs_lst)))
 
-    urls: dict[str, str] = get_url(pgs_id, args.build)
+    urls: dict[str, str] = get_url(pgs_id, args.build, pgsc_calc_info)
 
     for pgsid, url in urls.items():
         logger.debug(f"Downloading {pgsid} from {url}")
@@ -135,6 +139,8 @@ def _parse_args(args=None) -> argparse.Namespace:
     parser.add_argument('-o', '--outdir', dest='outdir', required=True,
                         default='scores/',
                         help='<Required> Output directory to store downloaded files')
+    parser.add_argument('-c', '--pgsc_calc', dest='pgsc_calc',
+                        help='<Optional> Provide information about downloading scoring files via pgsc_calc')
     parser.add_argument('-v', '--verbose', dest='verbose', action='store_true',
                         help='<Optional> Extra logging information')
     return parser.parse_args(args)

pgscatalog_utils/download/publication.py

@@ -6,10 +6,10 @@
 logger = logging.getLogger(__name__)
 
 
-def query_publication(pgp: str) -> list[str]:
+def query_publication(pgp: str, user_agent:str = None) -> list[str]:
     logger.debug("Querying PGS Catalog with publication PGP ID")
     api: str = f'/publication/{pgp}'
-    results_json = query_api(api)
+    results_json = query_api(api, user_agent)
 
     if results_json == {} or results_json == None:
         logger.critical(f"Bad response from PGS Catalog for EFO term: {pgp}")

pgscatalog_utils/download/score.py

@@ -4,17 +4,18 @@
 import jq
 import requests
 import time
+from pgscatalog_utils import __version__ as pgscatalog_utils_version
 
 logger = logging.getLogger(__name__)
 
 
-def get_url(pgs: list[str], build: str) -> dict[str, str]:
+def get_url(pgs: list[str], build: str, user_agent:str = None) -> dict[str, str]:
     pgs_result: list[str] = []
     url_result: list[str] = []
 
     for chunk in _chunker(pgs):
         try:
-            response = _parse_json_query(query_score(chunk), build)
+            response = _parse_json_query(query_score(chunk,user_agent), build)
             pgs_result = pgs_result + list(response.keys())
             url_result = url_result + list(response.values())
         except TypeError:
@@ -29,13 +30,17 @@ def get_url(pgs: list[str], build: str) -> dict[str, str]:
     return dict(zip(pgs_result, url_result))
 
 
-def query_api(api: str, retry:int = 0) -> dict:
+def query_api(api: str, user_agent:str = None, retry:int = 0) -> dict:
     max_retries = 5
     wait = 60
     results_json = None
     rest_url_root = 'https://www.pgscatalog.org/rest'
+    # Set pgscatalog_utils user agent if none provided
+    if not user_agent:
+        user_agent = 'pgscatalog_utils/'+pgscatalog_utils_version
     try:
-        r: requests.models.Response = requests.get(rest_url_root+api)
+        headers = {'User-Agent': user_agent}
+        r: requests.models.Response = requests.get(rest_url_root+api, headers=headers)
         r.raise_for_status()
         results_json = r.json()
     except requests.exceptions.HTTPError as e:
@@ -54,10 +59,10 @@ def query_api(api: str, retry:int = 0) -> dict:
     return results_json
 
 
-def query_score(pgs_id: list[str]) -> dict:
+def query_score(pgs_id: list[str], user_agent:str = None) -> dict:
     pgs: str = ','.join(pgs_id)
     api: str = f'/score/search?pgs_ids={pgs}'
-    results_json = query_api(api)
+    results_json = query_api(api, user_agent)
     return results_json
 
 

pgscatalog_utils/download/trait.py

@@ -6,10 +6,10 @@
 logger = logging.getLogger(__name__)
 
 
-def query_trait(trait: str) -> list[str]:
+def query_trait(trait: str, user_agent:str = None) -> list[str]:
     logger.debug(f"Querying PGS Catalog with trait {trait}")
     api: str = f'/trait/{trait}?include_children=1'
-    results_json = query_api(api)
+    results_json = query_api(api, user_agent)
 
     if results_json == {} or results_json == None:
         logger.critical(f"Bad response from PGS Catalog for EFO term: {trait}")