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

1"""Core database utilities.""" 

2 

3from typing import cast 

4 

5import gws 

6import gws.lib.crs 

7 

8 

9class Config(gws.Config): 

10 """Database configuration""" 

11 

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

13 """Database providers.""" 

14 

15 

16class Object(gws.DatabaseManager): 

17 def configure(self): 

18 self.providers = [] 

19 

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

21 try: 

22 self.create_provider(cfg) 

23 except Exception as exc: 

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

25 

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

27 

28 ## 

29 

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

31 pass 

32 

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

34 # @TODO deinit providers 

35 pass 

36 

37 ## 

38 

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

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

41 

42 # replace a provider with the same uid 

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

44 m[prov.uid] = prov 

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

46 

47 return prov 

48 

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

50 if uid: 

51 for p in self.providers: 

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

53 return p 

54 

55 elif ext_type: 

56 for p in self.providers: 

57 if p.extType == ext_type: 

58 return p 

59 

60 elif self.providers: 

61 return self.providers[0]