We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Use `pytest-randomly` plugin
https://github.com/pytest-dev/pytest-randomly

This plugin randomizes the order the tests are run in which prevents us from relying on their order to succeed. It also seeds various sources of randomness to improve reproducibility. For instance, if we encounter a test failure in CI due to randomness, we can run the tests locally with the `pytest --randomly-seed=<seed used in the CI test run>` to reproduce.

I've used this plugin before, and it has been helpful numerous times. I've already run into problems with Meltano's tests that are caused by their execution order. Using this plugin should force us to fix those issues, and prevent them from returning.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `noxfile.py`
Content:
```
1 """Nox configuration."""
2 
3 from __future__ import annotations
4 
5 import os
6 import sys
7 from pathlib import Path
8 from textwrap import dedent
9 
10 try:
11     from nox_poetry import Session
12     from nox_poetry import session as nox_session
13 except ImportError:
14     message = f"""\
15     Nox failed to import the 'nox-poetry' package.
16     Please install it using the following command:
17     {sys.executable} -m pip install nox-poetry"""
18     raise SystemExit(dedent(message)) from None
19 
20 
21 package = "meltano"
22 python_versions = ["3.10", "3.9", "3.8", "3.7"]
23 main_python_version = "3.9"
24 locations = "src", "tests", "noxfile.py"
25 
26 
27 @nox_session(python=python_versions)
28 def tests(session: Session) -> None:
29     """Execute pytest tests and compute coverage.
30 
31     Args:
32         session: Nox session.
33     """
34     backend_db = os.environ.get("PYTEST_BACKEND", "sqlite")
35 
36     if backend_db == "mssql":
37         session.install(".[mssql]")
38     else:
39         session.install(".")
40 
41     session.install(
42         "coverage[toml]",
43         "freezegun",
44         "mock",
45         "pytest",
46         "pytest-asyncio",
47         "pytest-docker",
48         "requests-mock",
49     )
50 
51     try:
52         session.run(
53             "coverage",
54             "run",
55             "--parallel",
56             "-m",
57             "pytest",
58             *session.posargs,
59             env={"NOX_CURRENT_SESSION": "tests"},
60         )
61     finally:
62         if session.interactive:
63             session.notify("coverage", posargs=[])
64 
65 
66 @nox_session(python=main_python_version)
67 def coverage(session: Session) -> None:
68     """Upload coverage data.
69 
70     Args:
71         session: Nox session.
72     """
73     args = session.posargs or ["report"]
74 
75     session.install("coverage[toml]")
76 
77     if not session.posargs and any(Path().glob(".coverage.*")):
78         session.run("coverage", "combine")
79 
80     session.run("coverage", *args)
81
```

Path: `src/meltano/core/db.py`
Content:
```
1 """Defines helpers related to the system database."""
2 
3 from __future__ import annotations
4 
5 import logging
6 import time
7 
8 from sqlalchemy import create_engine
9 from sqlalchemy.engine import Connection, Engine
10 from sqlalchemy.exc import OperationalError
11 from sqlalchemy.orm import sessionmaker
12 from sqlalchemy.sql import text
13 
14 from meltano.core.project import Project
15 
16 from .project_settings_service import ProjectSettingsService
17 
18 # Keep a Project → Engine mapping to serve
19 # the same engine for the same Project
20 _engines = {}
21 
22 
23 def project_engine(
24     project: Project,
25     default: bool = False,
26 ) -> tuple[Engine, sessionmaker]:
27     """Create and register a SQLAlchemy engine for a Meltano project instance.
28 
29     Args:
30         project: The Meltano project that the engine will be connected to.
31         default: Whether the engine created should be stored as the default
32             engine for this project.
33 
34     Returns:
35         The engine, and a session maker bound to the engine.
36     """
37     existing_engine = _engines.get(project)
38     if existing_engine:
39         return existing_engine
40 
41     settings = ProjectSettingsService(project)
42 
43     engine_uri = settings.get("database_uri")
44     logging.debug(f"Creating engine {project}@{engine_uri}")
45     engine = create_engine(engine_uri, pool_pre_ping=True)
46 
47     # Connect to the database to ensure it is available.
48     connect(
49         engine,
50         max_retries=settings.get("database_max_retries"),
51         retry_timeout=settings.get("database_retry_timeout"),
52     )
53 
54     init_hook(engine)
55 
56     engine_session = (engine, sessionmaker(bind=engine))
57 
58     if default:
59         # register the default engine
60         _engines[project] = engine_session
61 
62     return engine_session
63 
64 
65 def connect(
66     engine: Engine,
67     max_retries: int,
68     retry_timeout: float,
69 ) -> Connection:
70     """Connect to the database.
71 
72     Args:
73         engine: The DB engine with which the check will be performed.
74         max_retries: The maximum number of retries that will be attempted.
75         retry_timeout: The number of seconds to wait between retries.
76 
77     Raises:
78         OperationalError: Error during DB connection - max retries exceeded.
79 
80     Returns:
81         A connection to the database.
82     """
83     attempt = 0
84     while True:
85         try:
86             return engine.connect()
87         except OperationalError:
88             if attempt >= max_retries:
89                 logging.error(
90                     f"Could not connect to the database after {attempt} "
91                     "attempts. Max retries exceeded."
92                 )
93                 raise
94             attempt += 1
95             logging.info(
96                 f"DB connection failed. Will retry after {retry_timeout}s. "
97                 f"Attempt {attempt}/{max_retries}"
98             )
99             time.sleep(retry_timeout)
100 
101 
102 init_hooks = {
103     "sqlite": lambda x: x.execute("PRAGMA journal_mode=WAL"),
104 }
105 
106 
107 def init_hook(engine: Engine) -> None:
108     """Run the initialization hook for the provided DB engine.
109 
110     The initialization hooks are taken from the `meltano.core.db.init_hooks`
111     dictionary, which maps the dialect name of the engine to a unary function
112     which will be called with the provided DB engine.
113 
114     Args:
115         engine: The engine for which the init hook will be run.
116 
117     Raises:
118         Exception: The init hook raised an exception.
119     """
120     try:
121         hook = init_hooks[engine.dialect.name]
122     except KeyError:
123         return
124 
125     try:
126         hook(engine)
127     except Exception as ex:
128         raise Exception(f"Failed to initialize database: {ex!s}") from ex
129 
130 
131 def ensure_schema_exists(
132     engine: Engine,
133     schema_name: str,
134     grant_roles: tuple[str] = (),
135 ) -> None:
136     """Ensure the specified `schema_name` exists in the database.
137 
138     Args:
139         engine: The DB engine to be used.
140         schema_name: The name of the schema.
141         grant_roles: Roles to grant to the specified schema.
142     """
143     schema_identifier = schema_name
144     group_identifiers = ",".join(grant_roles)
145 
146     create_schema = text(f"CREATE SCHEMA IF NOT EXISTS {schema_identifier}")
147     grant_select_schema = text(
148         f"ALTER DEFAULT PRIVILEGES IN SCHEMA {schema_identifier} GRANT SELECT ON TABLES TO {group_identifiers}"
149     )
150     grant_usage_schema = text(
151         f"GRANT USAGE ON SCHEMA {schema_identifier} TO {group_identifiers}"
152     )
153 
154     with engine.connect() as conn, conn.begin():
155         conn.execute(create_schema)
156         if grant_roles:
157             conn.execute(grant_select_schema)
158             conn.execute(grant_usage_schema)
159 
160     logging.info(f"Schema {schema_name} has been created successfully.")
161     for role in grant_roles:
162         logging.info(f"Usage has been granted for role: {role}.")
163
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/noxfile.py b/noxfile.py
--- a/noxfile.py
+++ b/noxfile.py
@@ -5,6 +5,7 @@
 import os
 import sys
 from pathlib import Path
+from random import randint
 from textwrap import dedent
 
 try:
@@ -45,6 +46,8 @@
         "pytest",
         "pytest-asyncio",
         "pytest-docker",
+        "pytest-order",
+        "pytest-randomly",
         "requests-mock",
     )
 
@@ -55,6 +58,7 @@
             "--parallel",
             "-m",
             "pytest",
+            f"--randomly-seed={randint(0, 2**32-1)}",  # noqa: S311, WPS432
             *session.posargs,
             env={"NOX_CURRENT_SESSION": "tests"},
         )
diff --git a/src/meltano/core/db.py b/src/meltano/core/db.py
--- a/src/meltano/core/db.py
+++ b/src/meltano/core/db.py
@@ -9,6 +9,7 @@
 from sqlalchemy.engine import Connection, Engine
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import NullPool
 from sqlalchemy.sql import text
 
 from meltano.core.project import Project
@@ -41,8 +42,9 @@
     settings = ProjectSettingsService(project)
 
     engine_uri = settings.get("database_uri")
-    logging.debug(f"Creating engine {project}@{engine_uri}")
-    engine = create_engine(engine_uri, pool_pre_ping=True)
+    logging.debug(f"Creating engine '{project}@{engine_uri}'")
+
+    engine = create_engine(engine_uri, poolclass=NullPool)
 
     # Connect to the database to ensure it is available.
     connect(

{"golden_diff": "diff --git a/noxfile.py b/noxfile.py\n--- a/noxfile.py\n+++ b/noxfile.py\n@@ -5,6 +5,7 @@\n import os\n import sys\n from pathlib import Path\n+from random import randint\n from textwrap import dedent\n \n try:\n@@ -45,6 +46,8 @@\n         \"pytest\",\n         \"pytest-asyncio\",\n         \"pytest-docker\",\n+        \"pytest-order\",\n+        \"pytest-randomly\",\n         \"requests-mock\",\n     )\n \n@@ -55,6 +58,7 @@\n             \"--parallel\",\n             \"-m\",\n             \"pytest\",\n+            f\"--randomly-seed={randint(0, 2**32-1)}\",  # noqa: S311, WPS432\n             *session.posargs,\n             env={\"NOX_CURRENT_SESSION\": \"tests\"},\n         )\ndiff --git a/src/meltano/core/db.py b/src/meltano/core/db.py\n--- a/src/meltano/core/db.py\n+++ b/src/meltano/core/db.py\n@@ -9,6 +9,7 @@\n from sqlalchemy.engine import Connection, Engine\n from sqlalchemy.exc import OperationalError\n from sqlalchemy.orm import sessionmaker\n+from sqlalchemy.pool import NullPool\n from sqlalchemy.sql import text\n \n from meltano.core.project import Project\n@@ -41,8 +42,9 @@\n     settings = ProjectSettingsService(project)\n \n     engine_uri = settings.get(\"database_uri\")\n-    logging.debug(f\"Creating engine {project}@{engine_uri}\")\n-    engine = create_engine(engine_uri, pool_pre_ping=True)\n+    logging.debug(f\"Creating engine '{project}@{engine_uri}'\")\n+\n+    engine = create_engine(engine_uri, poolclass=NullPool)\n \n     # Connect to the database to ensure it is available.\n     connect(\n", "issue": "Use `pytest-randomly` plugin\nhttps://github.com/pytest-dev/pytest-randomly\r\n\r\nThis plugin randomizes the order the tests are run in which prevents us from relying on their order to succeed. It also seeds various sources of randomness to improve reproducibility. For instance, if we encounter a test failure in CI due to randomness, we can run the tests locally with the `pytest --randomly-seed=<seed used in the CI test run>` to reproduce.\r\n\r\nI've used this plugin before, and it has been helpful numerous times. I've already run into problems with Meltano's tests that are caused by their execution order. Using this plugin should force us to fix those issues, and prevent them from returning.\n", "before_files": [{"content": "\"\"\"Nox configuration.\"\"\"\n\nfrom __future__ import annotations\n\nimport os\nimport sys\nfrom pathlib import Path\nfrom textwrap import dedent\n\ntry:\n    from nox_poetry import Session\n    from nox_poetry import session as nox_session\nexcept ImportError:\n    message = f\"\"\"\\\n    Nox failed to import the 'nox-poetry' package.\n    Please install it using the following command:\n    {sys.executable} -m pip install nox-poetry\"\"\"\n    raise SystemExit(dedent(message)) from None\n\n\npackage = \"meltano\"\npython_versions = [\"3.10\", \"3.9\", \"3.8\", \"3.7\"]\nmain_python_version = \"3.9\"\nlocations = \"src\", \"tests\", \"noxfile.py\"\n\n\n@nox_session(python=python_versions)\ndef tests(session: Session) -> None:\n    \"\"\"Execute pytest tests and compute coverage.\n\n    Args:\n        session: Nox session.\n    \"\"\"\n    backend_db = os.environ.get(\"PYTEST_BACKEND\", \"sqlite\")\n\n    if backend_db == \"mssql\":\n        session.install(\".[mssql]\")\n    else:\n        session.install(\".\")\n\n    session.install(\n        \"coverage[toml]\",\n        \"freezegun\",\n        \"mock\",\n        \"pytest\",\n        \"pytest-asyncio\",\n        \"pytest-docker\",\n        \"requests-mock\",\n    )\n\n    try:\n        session.run(\n            \"coverage\",\n            \"run\",\n            \"--parallel\",\n            \"-m\",\n            \"pytest\",\n            *session.posargs,\n            env={\"NOX_CURRENT_SESSION\": \"tests\"},\n        )\n    finally:\n        if session.interactive:\n            session.notify(\"coverage\", posargs=[])\n\n\n@nox_session(python=main_python_version)\ndef coverage(session: Session) -> None:\n    \"\"\"Upload coverage data.\n\n    Args:\n        session: Nox session.\n    \"\"\"\n    args = session.posargs or [\"report\"]\n\n    session.install(\"coverage[toml]\")\n\n    if not session.posargs and any(Path().glob(\".coverage.*\")):\n        session.run(\"coverage\", \"combine\")\n\n    session.run(\"coverage\", *args)\n", "path": "noxfile.py"}, {"content": "\"\"\"Defines helpers related to the system database.\"\"\"\n\nfrom __future__ import annotations\n\nimport logging\nimport time\n\nfrom sqlalchemy import create_engine\nfrom sqlalchemy.engine import Connection, Engine\nfrom sqlalchemy.exc import OperationalError\nfrom sqlalchemy.orm import sessionmaker\nfrom sqlalchemy.sql import text\n\nfrom meltano.core.project import Project\n\nfrom .project_settings_service import ProjectSettingsService\n\n# Keep a Project \u2192 Engine mapping to serve\n# the same engine for the same Project\n_engines = {}\n\n\ndef project_engine(\n    project: Project,\n    default: bool = False,\n) -> tuple[Engine, sessionmaker]:\n    \"\"\"Create and register a SQLAlchemy engine for a Meltano project instance.\n\n    Args:\n        project: The Meltano project that the engine will be connected to.\n        default: Whether the engine created should be stored as the default\n            engine for this project.\n\n    Returns:\n        The engine, and a session maker bound to the engine.\n    \"\"\"\n    existing_engine = _engines.get(project)\n    if existing_engine:\n        return existing_engine\n\n    settings = ProjectSettingsService(project)\n\n    engine_uri = settings.get(\"database_uri\")\n    logging.debug(f\"Creating engine {project}@{engine_uri}\")\n    engine = create_engine(engine_uri, pool_pre_ping=True)\n\n    # Connect to the database to ensure it is available.\n    connect(\n        engine,\n        max_retries=settings.get(\"database_max_retries\"),\n        retry_timeout=settings.get(\"database_retry_timeout\"),\n    )\n\n    init_hook(engine)\n\n    engine_session = (engine, sessionmaker(bind=engine))\n\n    if default:\n        # register the default engine\n        _engines[project] = engine_session\n\n    return engine_session\n\n\ndef connect(\n    engine: Engine,\n    max_retries: int,\n    retry_timeout: float,\n) -> Connection:\n    \"\"\"Connect to the database.\n\n    Args:\n        engine: The DB engine with which the check will be performed.\n        max_retries: The maximum number of retries that will be attempted.\n        retry_timeout: The number of seconds to wait between retries.\n\n    Raises:\n        OperationalError: Error during DB connection - max retries exceeded.\n\n    Returns:\n        A connection to the database.\n    \"\"\"\n    attempt = 0\n    while True:\n        try:\n            return engine.connect()\n        except OperationalError:\n            if attempt >= max_retries:\n                logging.error(\n                    f\"Could not connect to the database after {attempt} \"\n                    \"attempts. Max retries exceeded.\"\n                )\n                raise\n            attempt += 1\n            logging.info(\n                f\"DB connection failed. Will retry after {retry_timeout}s. \"\n                f\"Attempt {attempt}/{max_retries}\"\n            )\n            time.sleep(retry_timeout)\n\n\ninit_hooks = {\n    \"sqlite\": lambda x: x.execute(\"PRAGMA journal_mode=WAL\"),\n}\n\n\ndef init_hook(engine: Engine) -> None:\n    \"\"\"Run the initialization hook for the provided DB engine.\n\n    The initialization hooks are taken from the `meltano.core.db.init_hooks`\n    dictionary, which maps the dialect name of the engine to a unary function\n    which will be called with the provided DB engine.\n\n    Args:\n        engine: The engine for which the init hook will be run.\n\n    Raises:\n        Exception: The init hook raised an exception.\n    \"\"\"\n    try:\n        hook = init_hooks[engine.dialect.name]\n    except KeyError:\n        return\n\n    try:\n        hook(engine)\n    except Exception as ex:\n        raise Exception(f\"Failed to initialize database: {ex!s}\") from ex\n\n\ndef ensure_schema_exists(\n    engine: Engine,\n    schema_name: str,\n    grant_roles: tuple[str] = (),\n) -> None:\n    \"\"\"Ensure the specified `schema_name` exists in the database.\n\n    Args:\n        engine: The DB engine to be used.\n        schema_name: The name of the schema.\n        grant_roles: Roles to grant to the specified schema.\n    \"\"\"\n    schema_identifier = schema_name\n    group_identifiers = \",\".join(grant_roles)\n\n    create_schema = text(f\"CREATE SCHEMA IF NOT EXISTS {schema_identifier}\")\n    grant_select_schema = text(\n        f\"ALTER DEFAULT PRIVILEGES IN SCHEMA {schema_identifier} GRANT SELECT ON TABLES TO {group_identifiers}\"\n    )\n    grant_usage_schema = text(\n        f\"GRANT USAGE ON SCHEMA {schema_identifier} TO {group_identifiers}\"\n    )\n\n    with engine.connect() as conn, conn.begin():\n        conn.execute(create_schema)\n        if grant_roles:\n            conn.execute(grant_select_schema)\n            conn.execute(grant_usage_schema)\n\n    logging.info(f\"Schema {schema_name} has been created successfully.\")\n    for role in grant_roles:\n        logging.info(f\"Usage has been granted for role: {role}.\")\n", "path": "src/meltano/core/db.py"}], "after_files": [{"content": "\"\"\"Nox configuration.\"\"\"\n\nfrom __future__ import annotations\n\nimport os\nimport sys\nfrom pathlib import Path\nfrom random import randint\nfrom textwrap import dedent\n\ntry:\n    from nox_poetry import Session\n    from nox_poetry import session as nox_session\nexcept ImportError:\n    message = f\"\"\"\\\n    Nox failed to import the 'nox-poetry' package.\n    Please install it using the following command:\n    {sys.executable} -m pip install nox-poetry\"\"\"\n    raise SystemExit(dedent(message)) from None\n\n\npackage = \"meltano\"\npython_versions = [\"3.10\", \"3.9\", \"3.8\", \"3.7\"]\nmain_python_version = \"3.9\"\nlocations = \"src\", \"tests\", \"noxfile.py\"\n\n\n@nox_session(python=python_versions)\ndef tests(session: Session) -> None:\n    \"\"\"Execute pytest tests and compute coverage.\n\n    Args:\n        session: Nox session.\n    \"\"\"\n    backend_db = os.environ.get(\"PYTEST_BACKEND\", \"sqlite\")\n\n    if backend_db == \"mssql\":\n        session.install(\".[mssql]\")\n    else:\n        session.install(\".\")\n\n    session.install(\n        \"coverage[toml]\",\n        \"freezegun\",\n        \"mock\",\n        \"pytest\",\n        \"pytest-asyncio\",\n        \"pytest-docker\",\n        \"pytest-order\",\n        \"pytest-randomly\",\n        \"requests-mock\",\n    )\n\n    try:\n        session.run(\n            \"coverage\",\n            \"run\",\n            \"--parallel\",\n            \"-m\",\n            \"pytest\",\n            f\"--randomly-seed={randint(0, 2**32-1)}\",  # noqa: S311, WPS432\n            *session.posargs,\n            env={\"NOX_CURRENT_SESSION\": \"tests\"},\n        )\n    finally:\n        if session.interactive:\n            session.notify(\"coverage\", posargs=[])\n\n\n@nox_session(python=main_python_version)\ndef coverage(session: Session) -> None:\n    \"\"\"Upload coverage data.\n\n    Args:\n        session: Nox session.\n    \"\"\"\n    args = session.posargs or [\"report\"]\n\n    session.install(\"coverage[toml]\")\n\n    if not session.posargs and any(Path().glob(\".coverage.*\")):\n        session.run(\"coverage\", \"combine\")\n\n    session.run(\"coverage\", *args)\n", "path": "noxfile.py"}, {"content": "\"\"\"Defines helpers related to the system database.\"\"\"\n\nfrom __future__ import annotations\n\nimport logging\nimport time\n\nfrom sqlalchemy import create_engine\nfrom sqlalchemy.engine import Connection, Engine\nfrom sqlalchemy.exc import OperationalError\nfrom sqlalchemy.orm import sessionmaker\nfrom sqlalchemy.pool import NullPool\nfrom sqlalchemy.sql import text\n\nfrom meltano.core.project import Project\n\nfrom .project_settings_service import ProjectSettingsService\n\n# Keep a Project \u2192 Engine mapping to serve\n# the same engine for the same Project\n_engines = {}\n\n\ndef project_engine(\n    project: Project,\n    default: bool = False,\n) -> tuple[Engine, sessionmaker]:\n    \"\"\"Create and register a SQLAlchemy engine for a Meltano project instance.\n\n    Args:\n        project: The Meltano project that the engine will be connected to.\n        default: Whether the engine created should be stored as the default\n            engine for this project.\n\n    Returns:\n        The engine, and a session maker bound to the engine.\n    \"\"\"\n    existing_engine = _engines.get(project)\n    if existing_engine:\n        return existing_engine\n\n    settings = ProjectSettingsService(project)\n\n    engine_uri = settings.get(\"database_uri\")\n    logging.debug(f\"Creating engine '{project}@{engine_uri}'\")\n\n    engine = create_engine(engine_uri, poolclass=NullPool)\n\n    # Connect to the database to ensure it is available.\n    connect(\n        engine,\n        max_retries=settings.get(\"database_max_retries\"),\n        retry_timeout=settings.get(\"database_retry_timeout\"),\n    )\n\n    init_hook(engine)\n\n    engine_session = (engine, sessionmaker(bind=engine))\n\n    if default:\n        # register the default engine\n        _engines[project] = engine_session\n\n    return engine_session\n\n\ndef connect(\n    engine: Engine,\n    max_retries: int,\n    retry_timeout: float,\n) -> Connection:\n    \"\"\"Connect to the database.\n\n    Args:\n        engine: The DB engine with which the check will be performed.\n        max_retries: The maximum number of retries that will be attempted.\n        retry_timeout: The number of seconds to wait between retries.\n\n    Raises:\n        OperationalError: Error during DB connection - max retries exceeded.\n\n    Returns:\n        A connection to the database.\n    \"\"\"\n    attempt = 0\n    while True:\n        try:\n            return engine.connect()\n        except OperationalError:\n            if attempt >= max_retries:\n                logging.error(\n                    f\"Could not connect to the database after {attempt} \"\n                    \"attempts. Max retries exceeded.\"\n                )\n                raise\n            attempt += 1\n            logging.info(\n                f\"DB connection failed. Will retry after {retry_timeout}s. \"\n                f\"Attempt {attempt}/{max_retries}\"\n            )\n            time.sleep(retry_timeout)\n\n\ninit_hooks = {\n    \"sqlite\": lambda x: x.execute(\"PRAGMA journal_mode=WAL\"),\n}\n\n\ndef init_hook(engine: Engine) -> None:\n    \"\"\"Run the initialization hook for the provided DB engine.\n\n    The initialization hooks are taken from the `meltano.core.db.init_hooks`\n    dictionary, which maps the dialect name of the engine to a unary function\n    which will be called with the provided DB engine.\n\n    Args:\n        engine: The engine for which the init hook will be run.\n\n    Raises:\n        Exception: The init hook raised an exception.\n    \"\"\"\n    try:\n        hook = init_hooks[engine.dialect.name]\n    except KeyError:\n        return\n\n    try:\n        hook(engine)\n    except Exception as ex:\n        raise Exception(f\"Failed to initialize database: {ex!s}\") from ex\n\n\ndef ensure_schema_exists(\n    engine: Engine,\n    schema_name: str,\n    grant_roles: tuple[str] = (),\n) -> None:\n    \"\"\"Ensure the specified `schema_name` exists in the database.\n\n    Args:\n        engine: The DB engine to be used.\n        schema_name: The name of the schema.\n        grant_roles: Roles to grant to the specified schema.\n    \"\"\"\n    schema_identifier = schema_name\n    group_identifiers = \",\".join(grant_roles)\n\n    create_schema = text(f\"CREATE SCHEMA IF NOT EXISTS {schema_identifier}\")\n    grant_select_schema = text(\n        f\"ALTER DEFAULT PRIVILEGES IN SCHEMA {schema_identifier} GRANT SELECT ON TABLES TO {group_identifiers}\"\n    )\n    grant_usage_schema = text(\n        f\"GRANT USAGE ON SCHEMA {schema_identifier} TO {group_identifiers}\"\n    )\n\n    with engine.connect() as conn, conn.begin():\n        conn.execute(create_schema)\n        if grant_roles:\n            conn.execute(grant_select_schema)\n            conn.execute(grant_usage_schema)\n\n    logging.info(f\"Schema {schema_name} has been created successfully.\")\n    for role in grant_roles:\n        logging.info(f\"Usage has been granted for role: {role}.\")\n", "path": "src/meltano/core/db.py"}]}