Returns a list of objects from the database. The kwargs parameter can contain any number of attributes. Only objects which contain all listed attributes and in which all values match for all listed attributes will be returned.
(self, **kwargs)
| 166 | session.close() |
| 167 | |
| 168 | def filter(self, **kwargs): |
| 169 | """ |
| 170 | Returns a list of objects from the database. |
| 171 | The kwargs parameter can contain any number |
| 172 | of attributes. Only objects which contain all |
| 173 | listed attributes and in which all values match |
| 174 | for all listed attributes will be returned. |
| 175 | """ |
| 176 | from sqlalchemy import or_ |
| 177 | |
| 178 | Statement = self.get_model('statement') |
| 179 | Tag = self.get_model('tag') |
| 180 | |
| 181 | page_size = kwargs.pop('page_size', 1000) |
| 182 | order_by = kwargs.pop('order_by', None) |
| 183 | tags = kwargs.pop('tags', []) |
| 184 | exclude_text = kwargs.pop('exclude_text', None) |
| 185 | exclude_text_words = kwargs.pop('exclude_text_words', []) |
| 186 | persona_not_startswith = kwargs.pop('persona_not_startswith', None) |
| 187 | search_text_contains = kwargs.pop('search_text_contains', None) |
| 188 | search_in_response_to_contains = kwargs.pop('search_in_response_to_contains', None) |
| 189 | |
| 190 | # Convert a single sting into a list if only one tag is provided |
| 191 | if isinstance(tags, str): |
| 192 | tags = [tags] |
| 193 | |
| 194 | # Use context manager to ensure session cleanup even if generator is partially consumed |
| 195 | session = self.Session() |
| 196 | try: |
| 197 | if len(kwargs) == 0: |
| 198 | statements = session.query(Statement).filter() |
| 199 | else: |
| 200 | statements = session.query(Statement).filter_by(**kwargs) |
| 201 | |
| 202 | if tags: |
| 203 | statements = statements.join(Statement.tags).filter( |
| 204 | Tag.name.in_(tags) |
| 205 | ) |
| 206 | |
| 207 | if exclude_text: |
| 208 | statements = statements.filter( |
| 209 | ~Statement.text.in_(exclude_text) |
| 210 | ) |
| 211 | |
| 212 | if exclude_text_words: |
| 213 | or_word_query = [ |
| 214 | Statement.text.ilike('%' + word + '%') for word in exclude_text_words |
| 215 | ] |
| 216 | statements = statements.filter( |
| 217 | ~or_(*or_word_query) |
| 218 | ) |
| 219 | |
| 220 | if persona_not_startswith: |
| 221 | statements = statements.filter( |
| 222 | ~Statement.persona.startswith('bot:') |
| 223 | ) |
| 224 | |
| 225 | if search_text_contains: |
no test coverage detected