Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/gitdb/db/ref.py: 17%

1# Copyright (C) 2010, 2011 Sebastian Thiel (byronimo@gmail.com) and contributors 

2# 

3# This module is part of GitDB and is released under 

4# the New BSD License: http://www.opensource.org/licenses/bsd-license.php 

5import codecs 

6from gitdb.db.base import ( 

7 CompoundDB, 

8) 

9 

10__all__ = ('ReferenceDB', ) 

11 

12 

13class ReferenceDB(CompoundDB): 

14 

15 """A database consisting of database referred to in a file""" 

16 

17 # Configuration 

18 # Specifies the object database to use for the paths found in the alternates 

19 # file. If None, it defaults to the GitDB 

20 ObjectDBCls = None 

21 

22 def __init__(self, ref_file): 

23 super().__init__() 

24 self._ref_file = ref_file 

25 

26 def _set_cache_(self, attr): 

27 if attr == '_dbs': 

28 self._dbs = list() 

29 self._update_dbs_from_ref_file() 

30 else: 

31 super()._set_cache_(attr) 

32 # END handle attrs 

33 

34 def _update_dbs_from_ref_file(self): 

35 dbcls = self.ObjectDBCls 

36 if dbcls is None: 

37 # late import 

38 from gitdb.db.git import GitDB 

39 dbcls = GitDB 

40 # END get db type 

41 

42 # try to get as many as possible, don't fail if some are unavailable 

43 ref_paths = list() 

44 try: 

45 with codecs.open(self._ref_file, 'r', encoding="utf-8") as f: 

46 ref_paths = [l.strip() for l in f] 

47 except OSError: 

48 pass 

49 # END handle alternates 

50 

51 ref_paths_set = set(ref_paths) 

52 cur_ref_paths_set = {db.root_path() for db in self._dbs} 

53 

54 # remove existing 

55 for path in (cur_ref_paths_set - ref_paths_set): 

56 for i, db in enumerate(self._dbs[:]): 

57 if db.root_path() == path: 

58 del(self._dbs[i]) 

59 continue 

60 # END del matching db 

61 # END for each path to remove 

62 

63 # add new 

64 # sort them to maintain order 

65 added_paths = sorted(ref_paths_set - cur_ref_paths_set, key=lambda p: ref_paths.index(p)) 

66 for path in added_paths: 

67 try: 

68 db = dbcls(path) 

69 # force an update to verify path 

70 if isinstance(db, CompoundDB): 

71 db.databases() 

72 # END verification 

73 self._dbs.append(db) 

74 except Exception: 

75 # ignore invalid paths or issues 

76 pass 

77 # END for each path to add 

78 

79 def update_cache(self, force=False): 

80 # re-read alternates and update databases 

81 self._update_dbs_from_ref_file() 

82 return super().update_cache(force)