Coverage for bbconf/bbagent.py: 59%

1from functools import cached_property 

2from pathlib import Path 

3from typing import Union, List 

4 

5from sqlalchemy.orm import Session 

6from sqlalchemy.sql import distinct, func, select 

7 

8from bbconf.config_parser.bedbaseconfig import BedBaseConfig 

9from bbconf.db_utils import Bed, BedSets, License 

10from bbconf.models.base_models import StatsReturn 

11from bbconf.modules.bedfiles import BedAgentBedFile 

12from bbconf.modules.bedsets import BedAgentBedSet 

13from bbconf.modules.objects import BBObjects 

14 

15 

16class BedBaseAgent(object): 

17 def __init__( 

18 self, 

19 config: Union[Path, str], 

20 ): 

21 """ 

22 Initialize connection to the pep_db database. You can use The basic connection parameters 

23 or libpq connection string. 

24 

25 """ 

26 

27 self.config = BedBaseConfig(config) 

28 

29 self._bed = BedAgentBedFile(self.config, self) 

30 self._bedset = BedAgentBedSet(self.config) 

31 self._objects = BBObjects(self.config) 

32 

33 @property 

34 def bed(self) -> BedAgentBedFile: 

35 return self._bed 

36 

37 @property 

38 def bedset(self) -> BedAgentBedSet: 

39 return self._bedset 

40 

41 @property 

42 def objects(self) -> BBObjects: 

43 return self._objects 

44 

45 @cached_property 

46 def get_stats(self) -> StatsReturn: 

47 """ 

48 Get statistics for a bed file 

49 

50 :return: statistics 

51 """ 

52 with Session(self.config.db_engine.engine) as session: 

53 number_of_bed = session.execute(select(func.count(Bed.id))).one()[0] 

54 number_of_bedset = session.execute(select(func.count(BedSets.id))).one()[0] 

55 

56 number_of_genomes = session.execute( 

57 select(func.count(distinct(Bed.genome_alias))) 

58 ).one()[0] 

59 

60 return StatsReturn( 

61 bedfiles_number=number_of_bed, 

62 bedsets_number=number_of_bedset, 

63 genomes_number=number_of_genomes, 

64 ) 

65 

66 @cached_property 

67 def list_of_licenses(self) -> List[str]: 

68 """ 

69 Get list of licenses from the database 

70 

71 :return: list of licenses 

72 """ 

73 statement = select(License.id) 

74 with Session(self.config.db_engine.engine) as session: 

75 licenses = session.execute(statement).all() 

76 return [result[0] for result in licenses]