Coverage for gws-app/gws/base/database/manager.py: 89%

1"""Database manager.""" 

2 

3from typing import cast 

4 

5import gws 

6import gws.lib.crs 

7 

8 

9class Config(gws.Config): 

10 """Database connections used by layers, models and other objects.""" 

11 

12 providers: list[gws.ext.config.databaseProvider] 

13 """Database connections.""" 

14 

15 

16class Object(gws.DatabaseManager): 

17 """Database manager.""" 

18 

19 def configure(self): 

20 self.providers = [] 

21 

22 for cfg in self.cfg('providers', default=[]): 

23 try: 

24 self.create_provider(cfg) 

25 except Exception as exc: 

26 gws.log.error(f'failed to create database provider {cfg}: {exc}') 

27 

28 self.root.app.middlewareMgr.register(self, 'db') 

29 

30 ## 

31 

32 def enter_middleware(self, req: gws.WebRequester): 

33 pass 

34 

35 def exit_middleware(self, req: gws.WebRequester, res: gws.WebResponder): 

36 # @TODO deinit providers 

37 pass 

38 

39 ## 

40 

41 def create_provider(self, cfg, **kwargs): 

42 prov = self.root.create_shared(gws.ext.object.databaseProvider, cfg, **kwargs) 

43 

44 # replace a provider with the same uid 

45 m = {p.uid: p for p in self.providers} 

46 m[prov.uid] = prov 

47 self.providers = list(m.values()) 

48 

49 return prov 

50 

51 def find_provider(self, uid=None, ext_type=None): 

52 if uid: 

53 for p in self.providers: 

54 if p.uid == uid and (not ext_type or p.extType == ext_type): 

55 return p 

56 

57 elif ext_type: 

58 for p in self.providers: 

59 if p.extType == ext_type: 

60 return p 

61 

62 elif self.providers: 

63 return self.providers[0]