(self, source_id, evfilter)
| 314 | return self._list_to_sql_str(fields, False) |
| 315 | |
| 316 | def _get_select_clauses(self, source_id, evfilter): |
| 317 | s_clauses = [] |
| 318 | for select_clause in evfilter.SelectClauses: |
| 319 | try: |
| 320 | if not select_clause.BrowsePath: |
| 321 | s_clauses.append(select_clause.Attribute.name) |
| 322 | else: |
| 323 | name = select_clause.BrowsePath[0].Name |
| 324 | s_clauses.append(name) |
| 325 | except AttributeError: |
| 326 | self.logger.warning('Historizing SQL OPC UA Select Clause Warning for node %s,' |
| 327 | ' Clause: %s:', source_id, select_clause) |
| 328 | |
| 329 | # remove select clauses that the event type doesn't have; SQL will error because the column doesn't exist |
| 330 | clauses = [x for x in s_clauses if x in self._event_fields[source_id]] |
| 331 | return clauses, self._list_to_sql_str(clauses) |
| 332 | |
| 333 | @staticmethod |
| 334 | def _list_to_sql_str(ls, quotes=True): |
no test coverage detected