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

1"""Core database utilities.""" 

2 

3from typing import cast 

4 

5import gws 

6import gws.gis.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 self.create_provider(cfg) 

22 

23 self.register_middleware('db') 

24 

25 ## 

26 

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

28 pass 

29 

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

31 # @TODO deinit providers 

32 pass 

33 

34 ## 

35 

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

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

38 

39 # replace a provider with the same uid 

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

41 m[prov.uid] = prov 

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

43 

44 return prov 

45 

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

47 if uid: 

48 for p in self.providers: 

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

50 return p 

51 

52 elif ext_type: 

53 for p in self.providers: 

54 if p.extType == ext_type: 

55 return p 

56 

57 elif self.providers: 

58 return self.providers[0]