Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy_utils/types/scalar_list.py: 75%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

20 statements  

1import sqlalchemy as sa 

2from sqlalchemy import types 

3 

4 

5class ScalarListException(Exception): 

6 pass 

7 

8 

9class ScalarListType(types.TypeDecorator): 

10 """ 

11 ScalarListType type provides convenient way for saving multiple scalar 

12 values in one column. ScalarListType works like list on python side and 

13 saves the result as comma-separated list in the database (custom separators 

14 can also be used). 

15 

16 Example :: 

17 

18 

19 from sqlalchemy_utils import ScalarListType 

20 

21 

22 class User(Base): 

23 __tablename__ = 'user' 

24 id = sa.Column(sa.Integer, autoincrement=True) 

25 hobbies = sa.Column(ScalarListType()) 

26 

27 

28 user = User() 

29 user.hobbies = ['football', 'ice_hockey'] 

30 session.commit() 

31 

32 

33 You can easily set up integer lists too: 

34 

35 :: 

36 

37 

38 from sqlalchemy_utils import ScalarListType 

39 

40 

41 class Player(Base): 

42 __tablename__ = 'player' 

43 id = sa.Column(sa.Integer, autoincrement=True) 

44 points = sa.Column(ScalarListType(int)) 

45 

46 

47 player = Player() 

48 player.points = [11, 12, 8, 80] 

49 session.commit() 

50 

51 

52 ScalarListType is always stored as text. To use an array field on 

53 PostgreSQL database use variant construct:: 

54 

55 from sqlalchemy_utils import ScalarListType 

56 

57 

58 class Player(Base): 

59 __tablename__ = 'player' 

60 id = sa.Column(sa.Integer, autoincrement=True) 

61 points = sa.Column( 

62 ARRAY(Integer).with_variant(ScalarListType(int), 'sqlite') 

63 ) 

64 

65 

66 """ 

67 

68 impl = sa.UnicodeText() 

69 

70 cache_ok = True 

71 

72 def __init__(self, coerce_func=str, separator=','): 

73 self.separator = str(separator) 

74 self.coerce_func = coerce_func 

75 

76 def process_bind_param(self, value, dialect): 

77 # Convert list of values to unicode separator-separated list 

78 # Example: [1, 2, 3, 4] -> '1, 2, 3, 4' 

79 if value is not None: 

80 if any(self.separator in str(item) for item in value): 

81 raise ScalarListException( 

82 "List values can't contain string '%s' (its being used as " 

83 'separator. If you wish for scalar list values to contain ' 

84 'these strings, use a different separator string.)' % self.separator 

85 ) 

86 return self.separator.join(map(str, value)) 

87 

88 def process_result_value(self, value, dialect): 

89 if value is not None: 

90 if value == '': 

91 return [] 

92 # coerce each value 

93 return list(map(self.coerce_func, value.split(self.separator)))