()
| 132 | |
| 133 | @contextmanager |
| 134 | def session_scope_threaded(): |
| 135 | import shared.database.input as input # TODO fix circular imports |
| 136 | """Provide a transactional scope around a series of operations. |
| 137 | |
| 138 | """ |
| 139 | |
| 140 | session = scoped_session_registry() |
| 141 | try: |
| 142 | yield session |
| 143 | session.commit() |
| 144 | except Exception as e: |
| 145 | session.rollback() |
| 146 | raise |
| 147 | finally: |
| 148 | scoped_session_registry.remove() |
no test coverage detected