Adds more database tuning settings for PostgreSQL (#17236)

docs/v3/develop/settings-ref.mdx

@@ -887,6 +887,30 @@ Controls the application_name field for connections opened from the connection p
 **Supported environment variables**:
 `PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_APPLICATION_NAME`
 
+### `statement_cache_size`
+Controls statement cache size for PostgreSQL connections. Setting this to 0 is required when using PgBouncer in transaction mode. Defaults to None.
+
+**Type**: `integer | None`
+
+**Default**: `None`
+
+**TOML dotted key path**: `server.database.sqlalchemy.connect_args.statement_cache_size`
+
+**Supported environment variables**:
+`PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_STATEMENT_CACHE_SIZE`
+
+### `prepared_statement_cache_size`
+Controls the size of the statement cache for PostgreSQL connections. When set to 0, statement caching is disabled. Defaults to None to use SQLAlchemy's default behavior.
+
+**Type**: `integer | None`
+
+**Default**: `None`
+
+**TOML dotted key path**: `server.database.sqlalchemy.connect_args.prepared_statement_cache_size`
+
+**Supported environment variables**:
+`PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_PREPARED_STATEMENT_CACHE_SIZE`
+
 ---
 ## SQLAlchemySettings
 Settings for controlling SQLAlchemy behavior; note that these settings only take effect when

schemas/settings.schema.json

@@ -757,6 +757,38 @@
                         "PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_APPLICATION_NAME"
                     ],
                     "title": "Application Name"
+                },
+                "statement_cache_size": {
+                    "anyOf": [
+                        {
+                            "type": "integer"
+                        },
+                        {
+                            "type": "null"
+                        }
+                    ],
+                    "default": null,
+                    "description": "Controls statement cache size for PostgreSQL connections. Setting this to 0 is required when using PgBouncer in transaction mode. Defaults to None.",
+                    "supported_environment_variables": [
+                        "PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_STATEMENT_CACHE_SIZE"
+                    ],
+                    "title": "Statement Cache Size"
+                },
+                "prepared_statement_cache_size": {
+                    "anyOf": [
+                        {
+                            "type": "integer"
+                        },
+                        {
+                            "type": "null"
+                        }
+                    ],
+                    "default": null,
+                    "description": "Controls the size of the statement cache for PostgreSQL connections. When set to 0, statement caching is disabled. Defaults to None to use SQLAlchemy's default behavior.",
+                    "supported_environment_variables": [
+                        "PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_PREPARED_STATEMENT_CACHE_SIZE"
+                    ],
+                    "title": "Prepared Statement Cache Size"
                 }
             },
             "title": "SQLAlchemyConnectArgsSettings",

src/prefect/server/database/configurations.py

@@ -121,6 +121,8 @@ def __init__(
         sqlalchemy_pool_size: Optional[int] = None,
         sqlalchemy_max_overflow: Optional[int] = None,
         connection_app_name: Optional[str] = None,
+        statement_cache_size: Optional[int] = None,
+        prepared_statement_cache_size: Optional[int] = None,
     ) -> None:
         self.connection_url = connection_url
         self.echo: bool = echo or PREFECT_API_DATABASE_ECHO.value()
@@ -140,6 +142,14 @@ def __init__(
             connection_app_name
             or get_current_settings().server.database.sqlalchemy.connect_args.application_name
         )
+        self.statement_cache_size: Optional[int] = (
+            statement_cache_size
+            or get_current_settings().server.database.sqlalchemy.connect_args.statement_cache_size
+        )
+        self.prepared_statement_cache_size: Optional[int] = (
+            prepared_statement_cache_size
+            or get_current_settings().server.database.sqlalchemy.connect_args.prepared_statement_cache_size
+        )
 
     def unique_key(self) -> tuple[Hashable, ...]:
         """
@@ -208,21 +218,28 @@ async def engine(self) -> AsyncEngine:
         if cache_key not in ENGINES:
             kwargs: dict[str, Any] = (
                 get_current_settings().server.database.sqlalchemy.model_dump(
-                    mode="json",
+                    mode="json", exclude={"connect_args"}
                 )
             )
-            connect_args: dict[str, Any] = kwargs.pop("connect_args")
-            app_name = connect_args.pop("application_name", None)
+            connect_args: dict[str, Any] = {}
 
             if self.timeout is not None:
                 connect_args["command_timeout"] = self.timeout
 
             if self.connection_timeout is not None:
                 connect_args["timeout"] = self.connection_timeout
 
-            if self.connection_app_name is not None or app_name is not None:
+            if self.statement_cache_size is not None:
+                connect_args["statement_cache_size"] = self.statement_cache_size
+
+            if self.prepared_statement_cache_size is not None:
+                connect_args["prepared_statement_cache_size"] = (
+                    self.prepared_statement_cache_size
+                )
+
+            if self.connection_app_name is not None:
                 connect_args["server_settings"] = dict(
-                    application_name=self.connection_app_name or app_name
+                    application_name=self.connection_app_name
                 )
 
             if connect_args:

src/prefect/settings/models/server/database.py

@@ -30,6 +30,20 @@ class SQLAlchemyConnectArgsSettings(PrefectBaseSettings):
         description="Controls the application_name field for connections opened from the connection pool when using a PostgreSQL database with the Prefect backend.",
     )
 
+    statement_cache_size: Optional[int] = Field(
+        default=None,
+        description="Controls statement cache size for PostgreSQL connections. Setting this to 0 is required when using PgBouncer in transaction mode. Defaults to None.",
+    )
+
+    prepared_statement_cache_size: Optional[int] = Field(
+        default=None,
+        description=(
+            "Controls the size of the statement cache for PostgreSQL connections. "
+            "When set to 0, statement caching is disabled. Defaults to None to use "
+            "SQLAlchemy's default behavior."
+        ),
+    )
+
 
 class SQLAlchemySettings(PrefectBaseSettings):
     """

tests/test_settings.py

@@ -323,6 +323,12 @@
     "PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_APPLICATION_NAME": {
         "test_value": "prefect"
     },
+    "PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_PREPARED_STATEMENT_CACHE_SIZE": {
+        "test_value": 1
+    },
+    "PREFECT_SERVER_DATABASE_SQLALCHEMY_CONNECT_ARGS_STATEMENT_CACHE_SIZE": {
+        "test_value": 1
+    },
     "PREFECT_SERVER_DATABASE_SQLALCHEMY_MAX_OVERFLOW": {"test_value": 10},
     "PREFECT_SERVER_DATABASE_SQLALCHEMY_POOL_RECYCLE": {"test_value": 10},
     "PREFECT_SERVER_DATABASE_SQLALCHEMY_POOL_SIZE": {"test_value": 10},