Coverage for src/python/ensembl/io/genomio/database/meta_getter.py: 21%

1# See the NOTICE file distributed with this work for additional information 

2# regarding copyright ownership. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); 

5# you may not use this file except in compliance with the License. 

6# You may obtain a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, 

12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

13# See the License for the specific language governing permissions and 

14# limitations under the License. 

15"""Connect to a core database and retrieve a meta_key:meta_value pair(s) and dump meta_key/value 

16pairs to stdout / JSON. 

17""" 

18 

19__all__ = ["get_meta_values"] 

20 

21import argparse 

22import logging 

23import json 

24from pathlib import PosixPath 

25from pathlib import Path 

26 

27from sqlalchemy.engine import URL 

28 

29import ensembl.io.genomio 

30from ensembl.utils.argparse import ArgumentParser 

31from ensembl.utils import StrPath 

32from ensembl.utils.logging import init_logging_with_args 

33from .dbconnection_lite import DBConnectionLite 

34 

35 

36def get_meta_values(db_url: URL, meta_keys: StrPath | list[str]) -> dict[str, str]: 

37 """Returns a set of meta values based on set of 1 or more input DB meta_keys. 

38 

39 Args: 

40 db_url: Target core database URL. 

41 meta_keys: File path with one meta key per line or list of meta keys. 

42 

43 """ 

44 db_name = db_url.database 

45 core_db = DBConnectionLite(db_url) 

46 query_meta_keys = [] 

47 unpopulated_meta_keys = [] 

48 meta_values_located = {} 

49 input_keys_count = 0 

50 meta_populated = False 

51 

52 # Check input type and populated query list 

53 if isinstance(meta_keys, PosixPath): 

54 with Path(meta_keys).open(mode="r", encoding="UTF-8") as fh: 

55 for line in fh.readlines(): 

56 meta_key = line.strip() 

57 query_meta_keys.append(meta_key) 

58 elif isinstance(meta_keys, list): 

59 query_meta_keys = meta_keys 

60 

61 # Loop over input meta_key(s) and query DB 

62 for meta_key in query_meta_keys: 

63 input_keys_count += 1 

64 meta_value = core_db.get_meta_value(f"{meta_key}") 

65 

66 if meta_value is not None: 

67 meta_values_located[f"{meta_key}"] = meta_value 

68 else: 

69 unpopulated_meta_keys.append(f"{meta_key}") 

70 logging.info(f"Meta query returned no entry on meta_key: '{meta_key}'") 

71 

72 # Now assess what meta info was recovered and dump to JSON 

73 total_queries_located = len(meta_values_located) 

74 if total_queries_located >= 1: 

75 meta_populated = True 

76 if total_queries_located < input_keys_count: 

77 logging.info(f"Missing meta_key(s)-> {unpopulated_meta_keys}") 

78 else: 

79 logging.warning("Zero input query meta_keys present/populated.") 

80 

81 if meta_populated: 

82 meta_values_located["database_name"] = f"{db_name}" 

83 print(json.dumps(meta_values_located, sort_keys=True, indent=2)) 

84 return meta_values_located 

85 return {} 

86 

87 

88def parse_args(arg_list: list[str] | None) -> argparse.Namespace: 

89 """Return a populated namespace with the arguments parsed from a list or from the command line. 

90 

91 Args: 

92 arg_list: List of arguments to parse. If `None`, grab them from the command line. 

93 

94 """ 

95 parser = ArgumentParser(description=__doc__) 

96 parser.add_server_arguments(include_database=True, help="server url and core database") 

97 parser.add_argument_src_path( 

98 "--meta_keys_list", help="Input File | List with >=2 meta_keys to query target database." 

99 ) 

100 parser.add_argument("--version", action="version", version=ensembl.io.genomio.__version__) 

101 parser.add_log_arguments(add_log_file=False) 

102 return parser.parse_args(arg_list) 

103 

104 

105def main(arg_list: list[str] | None = None) -> None: 

106 """Main script entry-point. 

107 

108 Args: 

109 arg_list: Arguments to parse passing list to parse_args(). 

110 """ 

111 args = parse_args(arg_list) 

112 init_logging_with_args(args) 

113 

114 _ = get_meta_values(db_url=args.url, meta_keys=args.meta_keys_list)