summary |
shortlog |
log |
commit | commitdiff |
tree
raw |
patch |
inline | side by side (from parent 1:
e3353de)
This is only used in one place.
return (int(version_parts[0]), int(version_parts[1]))
return (int(version_parts[0]), int(version_parts[1]))
- def extension_loaded(self, extension_name: str) -> bool:
- """ Return True if the hstore extension is loaded in the database.
- """
- with self.cursor() as cur:
- cur.execute('SELECT extname FROM pg_extension WHERE extname = %s', (extension_name, ))
- return cur.rowcount > 0
-
-
class ConnectionContext(ContextManager[Connection]):
""" Context manager of the connection that also provides direct access
to the underlying connection.
"""
connection: Connection
class ConnectionContext(ContextManager[Connection]):
""" Context manager of the connection that also provides direct access
to the underlying connection.
"""
connection: Connection
def connect(dsn: str) -> ConnectionContext:
""" Open a connection to the database using the specialised connection
factory. The returned object may be used in conjunction with 'with'.
def connect(dsn: str) -> ConnectionContext:
""" Open a connection to the database using the specialised connection
factory. The returned object may be used in conjunction with 'with'.
def _require_loaded(extension_name: str, conn: Connection) -> None:
""" Check that the given extension is loaded. """
def _require_loaded(extension_name: str, conn: Connection) -> None:
""" Check that the given extension is loaded. """
- if not conn.extension_loaded(extension_name):
- LOG.fatal('Required module %s is not loaded.', extension_name)
- raise UsageError(f'{extension_name} is not loaded.')
+ with conn.cursor() as cur:
+ cur.execute('SELECT * FROM pg_extension WHERE extname = %s', (extension_name, ))
+ if cur.rowcount <= 0:
+ LOG.fatal('Required module %s is not loaded.', extension_name)
+ raise UsageError(f'{extension_name} is not loaded.')
def check_existing_database_plugins(dsn: str) -> None:
def check_existing_database_plugins(dsn: str) -> None: