Coverage for .tox/coverage/lib/python3.11/site-packages/wuttjamaican/db/handler.py: 100%

1# -*- coding: utf-8; -*- 

2################################################################################ 

3# 

4# WuttJamaican -- Base package for Wutta Framework 

5# Copyright © 2024 Lance Edgar 

6# 

7# This file is part of Wutta Framework. 

8# 

9# Wutta Framework is free software: you can redistribute it and/or modify it 

10# under the terms of the GNU General Public License as published by the Free 

11# Software Foundation, either version 3 of the License, or (at your option) any 

12# later version. 

13# 

14# Wutta Framework is distributed in the hope that it will be useful, but 

15# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 

16# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for 

17# more details. 

18# 

19# You should have received a copy of the GNU General Public License along with 

20# Wutta Framework. If not, see <http://www.gnu.org/licenses/>. 

21# 

22################################################################################ 

23""" 

24Database Handler 

25""" 

26 

27import sqlalchemy as sa 

28 

29from wuttjamaican.app import GenericHandler 

30 

31 

32class DatabaseHandler(GenericHandler): 

33 """ 

34 Base class and default implementation for the :term:`db handler`. 

35 """ 

36 

37 def get_dialect(self, bind): 

38 """ """ 

39 return bind.url.get_dialect().name 

40 

41 def next_counter_value(self, session, key): 

42 """ 

43 Return the next counter value for the given key. 

44 

45 If the DB backend is PostgreSQL, then a proper "sequence" is 

46 used for the counter. 

47 

48 All other backends use a "fake" sequence by creating a 

49 dedicated table with auto-increment primary key, to provide 

50 the counter. 

51 

52 :param session: Current :term:`db session`. 

53 

54 :param key: Unique key indicating the counter for which the 

55 next value should be fetched. 

56 

57 :returns: Next value as integer. 

58 """ 

59 dialect = self.get_dialect(session.bind) 

60 

61 # postgres uses "true" native sequence 

62 if dialect == 'postgresql': 

63 sql = f"create sequence if not exists {key}_seq" 

64 session.execute(sa.text(sql)) 

65 sql = f"select nextval('{key}_seq')" 

66 value = session.execute(sa.text(sql)).scalar() 

67 return value 

68 

69 # otherwise use "magic" workaround 

70 engine = session.bind 

71 metadata = sa.MetaData() 

72 table = sa.Table(f'_counter_{key}', metadata, 

73 sa.Column('value', sa.Integer(), primary_key=True)) 

74 table.create(engine, checkfirst=True) 

75 with engine.begin() as cxn: 

76 result = cxn.execute(table.insert()) 

77 return result.lastrowid