Add a small wrapper for APSW use (#3202)

* Add a small wrapper for APSW use

* changelog
This commit is contained in:
Michael H 2019-12-22 13:18:31 -05:00 committed by GitHub
parent b72c05d3d4
commit bf6297aaf2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 63 additions and 0 deletions

1
.github/CODEOWNERS vendored
View File

@ -25,6 +25,7 @@ redbot/core/utils/antispam.py @mikeshardmind
redbot/core/utils/tunnel.py @mikeshardmind redbot/core/utils/tunnel.py @mikeshardmind
redbot/core/utils/caching.py @mikeshardmind redbot/core/utils/caching.py @mikeshardmind
redbot/core/utils/common_filters.py @mikeshardmind redbot/core/utils/common_filters.py @mikeshardmind
redbot/core/utils/dbtools.py @mikeshardmind
# Cogs # Cogs
redbot/cogs/admin/* @tekulvw redbot/cogs/admin/* @tekulvw

View File

@ -0,0 +1 @@
Adds a not fully documented class for APSW interaction

View File

@ -0,0 +1,61 @@
from __future__ import annotations
from contextlib import contextmanager
from pathlib import Path
from typing import Generator, Union
import apsw
__all__ = ["APSWConnectionWrapper"]
# TODO (mikeshardmind): make this inherit typing_extensions.Protocol
# long term: mypy; short term: removing the pylint disables below
class ProvidesCursor:
def cursor(self) -> apsw.Cursor:
...
class ContextManagerMixin(ProvidesCursor):
@contextmanager
def with_cursor(self) -> Generator[apsw.Cursor, None, None]:
"""
apsw cursors are relatively cheap, and are gc safe
In most cases, it's fine not to use this.
"""
c = self.cursor() # pylint: disable=assignment-from-no-return
try:
yield c
finally:
c.close()
@contextmanager
def transaction(self) -> Generator[apsw.Cursor, None, None]:
"""
Wraps a cursor as a context manager for a transaction
which is rolled back on unhandled exception,
or commited on non-exception exit
"""
c = self.cursor() # pylint: disable=assignment-from-no-return
try:
c.execute("BEGIN TRANSACTION")
yield c
except Exception:
c.execute("ROLLBACK TRANSACTION")
raise
else:
c.execute("COMMIT TRANSACTION")
finally:
c.close()
class APSWConnectionWrapper(apsw.Connection, ContextManagerMixin):
"""
Provides a few convenience methods, and allows a path object for construction
"""
def __init__(self, filename: Union[Path, str], *args, **kwargs):
super().__init__(str(filename), *args, **kwargs)
# TODO (mikeshardmind): asyncio friendly ThreadedAPSWConnection class