Coverage for src/python/ensembl/io/genomio/database/core_server.py: 100%

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"""Interface to a Mysql server with core databases.""" 

16 

17__all__ = ["CoreServer"] 

18 

19import re 

20from typing import List, Optional 

21import logging 

22 

23import sqlalchemy 

24from sqlalchemy.engine import URL 

25from sqlalchemy import text 

26 

27 

28class CoreServer: 

29 """Basic interface to a MySQL server with core databases. 

30 

31 Allows to get a filtered list of databases. 

32 """ 

33 

34 def __init__(self, server_url: URL) -> None: 

35 logging.debug(f"Connect to {server_url}") 

36 self.engine = sqlalchemy.create_engine(server_url) 

37 

38 def get_all_core_names(self) -> List[str]: 

39 """Query the server and retrieve all database names that look like Ensembl cores.""" 

40 

41 with self.engine.connect() as connection: 

42 all_query = connection.execute(text(r"SHOW DATABASES LIKE '%%_core_%%'")) 

43 dbs = [row[0] for row in all_query.fetchall()] 

44 logging.info(f"{len(dbs)} core databases on the server") 

45 return dbs 

46 

47 def get_cores( 

48 self, 

49 *, 

50 prefix: str = "", 

51 build: Optional[int] = None, 

52 version: Optional[int] = None, 

53 dbname_re: str = "", 

54 db_list: Optional[List[str]] = None, 

55 ) -> List[str]: 

56 """Returns a list of core databases, filtered if requested. 

57 

58 Args: 

59 prefix: Filter by prefix (no "_" is added automatically). 

60 build: Filter by VEuPathDB build number. 

61 version: Filter by Ensembl version. 

62 dbname_re: Filter by dbname regular expression. 

63 db_list: Explicit list of database names. 

64 """ 

65 dbs = [] 

66 

67 dbs = self.get_all_core_names() 

68 

69 # Check if there are databases returned from query to host 

70 if not dbs: 

71 logging.warning("No databases returned from query") 

72 

73 if db_list: 

74 logging.debug(f"Filter with db list: {db_list}") 

75 dbs = [db for db in dbs if db in db_list] 

76 if prefix: 

77 dbs = [db for db in dbs if db.startswith(f"{prefix}")] 

78 if dbname_re: 

79 dbname_m = re.compile(dbname_re) 

80 dbs = list(filter(dbname_m.search, dbs)) 

81 if build is not None: 

82 dbs = [db for db in dbs if re.search(rf"_core_{build}_\d+_\d+$", db)] 

83 if version is not None: 

84 dbs = [db for db in dbs if re.search(rf"_core_\d+_{version}_\d+$", db)] 

85 

86 logging.info(f"{len(dbs)} core databases remain after filtering") 

87 

88 return dbs