Coverage for src/python/ensembl/io/genomio/database/dbconnection_lite.py: 96%

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"""Simplified Database interface to an Ensembl database, to make access to metadata easier and faster.""" 

16 

17__all__ = ["DBConnectionLite"] 

18 

19import logging 

20import re 

21from typing import Any, Dict, List, Optional 

22 

23from sqlalchemy import select 

24from sqlalchemy.orm import Session 

25 

26from ensembl.core.models import Meta 

27from ensembl.utils.database import DBConnection, StrURL 

28 

29_DB_PATTERN_RELEASE = re.compile(r".+_(?:core|otherfeatures|variation)_(?P<release>\d+)_\d+_\d+") 

30 

31 

32class DBConnectionLite(DBConnection): 

33 """Extension to get metadata directly from a database, assuming it has a metadata table.""" 

34 

35 def __init__(self, url: StrURL, reflect: bool = False, **kwargs: Any) -> None: 

36 super().__init__(url, reflect, **kwargs) 

37 self._metadata: Dict[str, List] = {} 

38 

39 def get_metadata(self) -> Dict[str, List]: 

40 """Retrieves all metadata from the `meta` table in the database. 

41 

42 Returns: 

43 A dict of with key meta_key, and value=List of meta_value. 

44 

45 """ 

46 self._load_metadata() 

47 return self._metadata 

48 

49 def _load_metadata(self) -> None: 

50 """Caches the metadata values.""" 

51 

52 if self._metadata: 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 return 

54 

55 with Session(self._engine) as session: 

56 meta_stmt = select(Meta) 

57 

58 for meta_row in session.scalars(meta_stmt).unique().all(): 

59 meta_key = meta_row.meta_key 

60 meta_value = meta_row.meta_value 

61 if meta_key in self._metadata: 

62 self._metadata[meta_key].append(meta_value) 

63 else: 

64 self._metadata[meta_key] = [meta_value] 

65 

66 def get_meta_value(self, meta_key: str) -> Optional[str]: 

67 """Returns the first meta_value for a given meta_key.""" 

68 

69 self._load_metadata() 

70 try: 

71 return self._metadata[meta_key][0] 

72 except KeyError: 

73 logging.debug(f"No meta_key {meta_key}") 

74 return None 

75 

76 def get_project_release(self) -> str: 

77 """Returns the project release number from the database name. Returns empty string if not found.""" 

78 

79 match = re.search(_DB_PATTERN_RELEASE, self.db_name) 

80 if match: 

81 return match.group(1) 

82 return ""