Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/cardinal_pythonlib/sqlalchemy/list_types.py : 56%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
2# cardinal_pythonlib/sqlalchemy/list_types.py
4"""
5===============================================================================
7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com).
9 This file is part of cardinal_pythonlib.
11 Licensed under the Apache License, Version 2.0 (the "License");
12 you may not use this file except in compliance with the License.
13 You may obtain a copy of the License at
15 https://www.apache.org/licenses/LICENSE-2.0
17 Unless required by applicable law or agreed to in writing, software
18 distributed under the License is distributed on an "AS IS" BASIS,
19 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 See the License for the specific language governing permissions and
21 limitations under the License.
23===============================================================================
25**SQLAlchemy type classes to store different kinds of lists in a database.**
27"""
29import csv
30from io import StringIO
31from typing import List, Optional
33from cardinal_pythonlib.logs import get_brace_style_log_with_null_handler
34from sqlalchemy.engine.interfaces import Dialect
35from sqlalchemy.sql.sqltypes import Text, UnicodeText
36from sqlalchemy.sql.type_api import TypeDecorator
38log = get_brace_style_log_with_null_handler(__name__)
41# =============================================================================
42# StringListType
43# =============================================================================
45class StringListType(TypeDecorator):
46 r"""
47 Store a list of strings as CSV.
48 (Rather less arbitrary in its encoding requirements than e.g.
49 https://sqlalchemy-utils.readthedocs.io/en/latest/_modules/sqlalchemy_utils/types/scalar_list.html#ScalarListType.)
51 - 2019-01-01: removed trailing ``\r\n`` (via ``lineterminator=""``).
53 Some related test code:
55 .. code-block:: python
57 import csv
58 from io import StringIO
60 pythonlist = [None, 1, "string", "commas, within string", "line 1\nline2"]
62 output_1 = StringIO()
63 wr_1 = csv.writer(output_1, quoting=csv.QUOTE_ALL) # appends '\r\n'
64 wr_1.writerow(pythonlist)
65 csvstring_1 = output_1.getvalue()
66 print(repr(csvstring_1))
67 backtopython_1 = list(csv.reader([csvstring_1]))[0]
68 print(repr(backtopython_1))
70 output_2 = StringIO()
71 wr_2 = csv.writer(output_2, quoting=csv.QUOTE_ALL, lineterminator="")
72 wr_2.writerow(pythonlist)
73 csvstring_2 = output_2.getvalue()
74 print(repr(csvstring_2))
75 backtopython_2 = list(csv.reader([csvstring_2]))[0]
76 print(repr(backtopython_2))
78 assert len(csvstring_1) > len(csvstring_2)
79 assert backtopython_1 == backtopython_2
81 So:
83 - The newline terminator is obviously unnecessary for something that will
84 always be a single CSV line.
85 - Eliminating it saves two bytes and adds clarity in the database
86 representation.
87 - Eliminating it keeps the system back-compatible, since the reader
88 happily reads things without the line terminator.
90 - **NOTE** in particular that this does not behave completely like a plain
91 Python list on the Python side, as follows.
93 - When an ORM object is created, the default value on the Python side is
94 ``None``.
96 - The SQLAlchemy ``default`` option is invoked at ``INSERT``, not at ORM
97 object creation; see
98 https://docs.sqlalchemy.org/en/latest/core/metadata.html#sqlalchemy.schema.Column.params.default.
100 - The SQLAlchemy ``server_default`` is the DDL ``DEFAULT`` value, not a
101 Python default.
103 - On database load, everything is fine (as ``process_result_value`` will
104 be called, which can translate a database ``NULL`` to a Python ``[]``).
106 - So that means that **if you want the field to be a list rather than
107 None from the outset,** you must set it to ``[]`` from ``__init__()``.
109 - Secondly, SQLAlchemy makes its columns behave in a special way **upon
110 assignment**. So, in particular, ``mylist.append(value)`` will not itself
111 mark the field as "dirty" and in need of writing to the database.
113 - Internally, support we define (on the class) ``mycol =
114 Column(Integer)``, and then create an instance via ``instance =
115 cls()``.
117 - Then ``cls.mycol`` will actually be of type
118 :class:`sqlalchemy.orm.attributes.InstrumentedAttribute`, and
119 ``instance.mycol`` will be of type ``int`` (or ``NoneType`` if it's
120 ``None``).
122 .. code-block:: python
124 from sqlalchemy.ext.declarative import declarative_base
125 from sqlalchemy.sql.schema import Column
126 from sqlalchemy.sql.sqltypes import Integer
128 Base = declarative_base()
130 class MyClass(Base):
131 __tablename__ = "mytable"
132 pk = Column(Integer, primary_key=True)
133 mycol = Column(Integer)
135 instance = MyClass()
136 type(MyClass.pk) # <class 'sqlalchemy.orm.attributes.InstrumentedAttribute'>
137 type(instance.pk) # <class 'NoneType'>
139 - The class :class:`sqlalchemy.orm.attributes.InstrumentedAttribute`
140 implements :meth:`__set__`, :meth:`__delete__`, and :meth:`__get__`.
141 This means that when you write ``instance.mycol = 5``, it calls the
142 ``__set__()`` function; see
143 https://docs.python.org/3.7/howto/descriptor.html.
145 - So, for a list (e.g. ``mylist = Column(StringListType)``, if you write
146 ``mylist = [value1, value2]``, it will call the appropriate
147 ``__set__()`` function and mark the field as "dirty" (see e.g.
148 :meth:`sqlalchemy.orm.attributes.ScalarAttributeImpl.set`). **But** if
149 ``mylist`` is already a list and you write ``mylist.append(value)``,
150 the ``__set__()`` function won't be called.
152 - If you haven't yet written the instance to the database, this doesn't
153 matter; "new" values are considered dirty and are written to the
154 database fine. But if you (a) create, (b) save, and then (c) append to
155 a list, the change won't be noticed. Since SQLAlchemy can save objects
156 for you as soon as another object needs to know it's PK, the fact that
157 (b) has happened may not be obvious.
159 - Therefore, in short, **beware append() and use assignment** for these
160 sorts of lists, if this might apply; e.g. ``mylist = mylist +
161 [value]``.
163 - Don't use ``+=``, either; that calls ``list.__iadd__()`` and modifies
164 the existing list, rather than calling
165 ``InstrumentedAttribute.__set__()``.
167 - So one method is to ignore ``__init__()`` (meaning new instances will
168 have the list-type field set to ``None``) and then using this sort of
169 access function:
171 .. code-block:: python
173 def add_to_mylist(self, text: str) -> None:
174 if self.mylist is None:
175 self.mylist = [text]
176 else:
177 # noinspection PyAugmentAssignment
178 self.mylist = self.mylist + [text] # not "append()", not "+="
180 """ # noqa
181 impl = UnicodeText()
183 @property
184 def python_type(self):
185 return list
187 @staticmethod
188 def _strlist_to_dbstr(strlist: Optional[List[str]]) -> str:
189 if not strlist:
190 return ""
191 output = StringIO()
192 wr = csv.writer(output, quoting=csv.QUOTE_ALL, lineterminator="")
193 wr.writerow(strlist)
194 return output.getvalue()
196 @staticmethod
197 def _dbstr_to_strlist(dbstr: Optional[str]) -> List[str]:
198 if not dbstr:
199 return []
200 try:
201 return list(csv.reader([dbstr]))[0]
202 # ... list( generator( list_of_lines ) )[first_line]
203 except csv.Error:
204 log.warning("StringListType: Unable to convert database value of "
205 "{!r} to Python; returning empty list", dbstr)
206 return []
208 def process_bind_param(self, value: Optional[List[str]],
209 dialect: Dialect) -> str:
210 """Convert things on the way from Python to the database."""
211 retval = self._strlist_to_dbstr(value)
212 return retval
214 def process_literal_param(self, value: Optional[List[str]],
215 dialect: Dialect) -> str:
216 """Convert things on the way from Python to the database."""
217 retval = self._strlist_to_dbstr(value)
218 return retval
220 # Could also use "process_literal_param = process_bind_param"
221 # or vice versa, but this adds some clarity via docstrings.
223 def process_result_value(self, value: Optional[str],
224 dialect: Dialect) -> List[str]:
225 """Convert things on the way from the database to Python."""
226 retval = self._dbstr_to_strlist(value)
227 return retval
230# =============================================================================
231# IntListType
232# =============================================================================
234class IntListType(TypeDecorator):
235 """
236 Store a list of integers as CSV.
238 **Note:** see :class:`StringListType` for a general discussion about
239 SQLAlchemy types where the Python representation is a list; they can seem
240 slightly unusual.
241 """
242 impl = Text()
244 @property
245 def python_type(self):
246 return list
248 @staticmethod
249 def _intlist_to_dbstr(intlist: Optional[List[int]]) -> str:
250 if not intlist:
251 return ""
252 return ",".join(str(x) for x in intlist)
254 @staticmethod
255 def _dbstr_to_intlist(dbstr: Optional[str]) -> List[int]:
256 if not dbstr:
257 return []
258 try:
259 return [int(x) for x in dbstr.split(",")]
260 except (TypeError, ValueError):
261 log.warning("IntListType: Unable to convert database value of {!r}"
262 " to Python; returning empty list", dbstr)
263 return []
265 def process_bind_param(self, value: Optional[List[int]],
266 dialect: Dialect) -> str:
267 """Convert things on the way from Python to the database."""
268 retval = self._intlist_to_dbstr(value)
269 return retval
271 def process_literal_param(self, value: Optional[List[int]],
272 dialect: Dialect) -> str:
273 """Convert things on the way from Python to the database."""
274 retval = self._intlist_to_dbstr(value)
275 return retval
277 def process_result_value(self, value: Optional[str],
278 dialect: Dialect) -> List[int]:
279 """Convert things on the way from the database to Python."""
280 retval = self._dbstr_to_intlist(value)
281 return retval