(self, get_connection, owner, database_name='', close_connection=None)
| 30 | |
| 31 | class SqliteSoxStorage: |
| 32 | def __init__(self, get_connection, owner, database_name='', close_connection=None): |
| 33 | self.get_connection = get_connection |
| 34 | self.close_connection = close_connection |
| 35 | self.owner = owner |
| 36 | self.database_name = database_name |
| 37 | |
| 38 | metadata_schema = sqlite_dao.TableSchema.load_schema(Metadata, [], owner_column='account_uid') |
| 39 | user_schema = sqlite_dao.TableSchema.load_schema(StorageUser, 'user_uid') |
| 40 | record_schema = sqlite_dao.TableSchema.load_schema(StorageRecord, 'record_uid') |
| 41 | record_aging_schema = sqlite_dao.TableSchema.load_schema(StorageRecordAging, 'record_uid') |
| 42 | user_record_schema = sqlite_dao.TableSchema.load_schema(StorageUserRecordLink, ['record_uid', 'user_uid'], |
| 43 | indexes={'UserUID': 'user_uid'}) |
| 44 | team_schema = sqlite_dao.TableSchema.load_schema(StorageTeam, 'team_uid') |
| 45 | team_user_schema = sqlite_dao.TableSchema.load_schema(StorageTeamUserLink, ['team_uid', 'user_uid'], |
| 46 | indexes={'UserUID': 'user_uid'}) |
| 47 | role_schema = sqlite_dao.TableSchema.load_schema(StorageRole, 'role_id') |
| 48 | record_permissions_schema = sqlite_dao.TableSchema.load_schema(StorageRecordPermissions, |
| 49 | ['record_uid', 'user_uid'], |
| 50 | indexes={'UserUID': 'user_uid'}) |
| 51 | shared_folder_record_schema = sqlite_dao.TableSchema.load_schema(StorageSharedFolderRecordLink, |
| 52 | ['folder_uid', 'record_uid'], |
| 53 | indexes={'RecordUID': 'record_uid'}) |
| 54 | shared_folder_user_schema = sqlite_dao.TableSchema.load_schema(StorageSharedFolderUserLink, |
| 55 | ['folder_uid', 'user_uid'], |
| 56 | indexes={'UserUID': 'user_uid'}) |
| 57 | shared_folder_team_schema = sqlite_dao.TableSchema.load_schema(StorageSharedFolderTeamLink, |
| 58 | ['folder_uid', 'team_uid'], |
| 59 | indexes={'TeamUID': 'team_uid'}) |
| 60 | sqlite_dao.verify_database( |
| 61 | self.get_connection(), |
| 62 | (metadata_schema, user_schema, record_schema, record_aging_schema, user_record_schema, team_schema, |
| 63 | team_user_schema, role_schema, record_permissions_schema, shared_folder_record_schema, |
| 64 | shared_folder_user_schema, shared_folder_team_schema) |
| 65 | ) |
| 66 | |
| 67 | self._metadata = sqlite.SqliteRecordStorage(self.get_connection, metadata_schema, owner) |
| 68 | self._users = sqlite.SqliteEntityStorage(self.get_connection, user_schema) |
| 69 | self._records = sqlite.SqliteEntityStorage(self.get_connection, record_schema) |
| 70 | self._record_aging = sqlite.SqliteEntityStorage(self.get_connection, record_aging_schema) |
| 71 | self._user_record_links = sqlite.SqliteLinkStorage(self.get_connection, user_record_schema) |
| 72 | self._teams = sqlite.SqliteEntityStorage(self.get_connection, team_schema) |
| 73 | self._team_user_links = sqlite.SqliteLinkStorage(self.get_connection, team_user_schema) |
| 74 | self._roles = sqlite.SqliteEntityStorage(self.get_connection, role_schema) |
| 75 | self._record_permissions = sqlite.SqliteLinkStorage(self.get_connection, record_permissions_schema) |
| 76 | self._sf_record_links = sqlite.SqliteLinkStorage(self.get_connection, shared_folder_record_schema) |
| 77 | self._sf_user_links = sqlite.SqliteLinkStorage(self.get_connection, shared_folder_user_schema) |
| 78 | self._sf_team_links = sqlite.SqliteLinkStorage(self.get_connection, shared_folder_team_schema) |
| 79 | |
| 80 | def _get_history(self): |
| 81 | return self._metadata.load() or Metadata() |
nothing calls this directly
no test coverage detected