diff --git a/README.md b/README.md
index 2e029641..566e8f98 100644
--- a/README.md
+++ b/README.md
@@ -48,7 +48,7 @@ pip install mangum[redis]==0.9.0b1
 
 ## Usage
 
-The `Mangum` adapter class is designed to wrap any ASGI application, accepting various configuration options, returning a callable. It can wrap an application and be assigned to the handler:
+The `Mangum` adapter class is designed to wrap any ASGI application and returns a callable. It can wrap an application and be assigned to the handler:
 
 ```python
 from mangum import Mangum
@@ -71,6 +71,55 @@ def handler(event, context):
     return response
 ```
 
+## Configuration
+
+The adapter accepts various arguments for configuring lifespan, logging, HTTP, WebSocket, and API Gateway behaviour.
+
+### Usage
+
+```python
+handler = Mangum(
+    app,
+    enable_lifespan=True,
+    log_level="info",
+    api_gateway_base_path=None,
+    text_mime_types=None,
+    dsn=None,
+    api_gateway_endpoint_url=None,
+    api_gateway_region_name=None
+)
+```
+
+#### Parameters
+
+- `enable_lifespan` : **bool**
+    
+    Specify whether or not to enable lifespan support. The adapter will automatically determine if lifespan is supported by the framework unless explicitly disabled.
+
+- `log_level` : **str**
+    
+    Level parameter for the logger.
+
+- `api_gateway_base_path` : **str**
+    
+    Base path to strip from URL when using a custom domain name.
+
+- `text_mime_types` : **list**
+        
+    The list of MIME types (in addition to the defaults) that should not return binary responses in API Gateway.
+
+- `dsn`: **str*
+    
+    DSN connection string to configure a supported WebSocket backend.
+
+- `api_gateway_endpoint_url` : **str**
+    
+    The endpoint url to use when sending data to WebSocket connections in API Gateway. This is useful if you are debugging locally with a package such as [serverless-dynamodb-local](https://github.com/99xt/serverless-dynamodb-local).
+
+- `api_gateway_region_name` : **str**
+    
+    The region name of the API Gateway that is managing the API connections.
+
 ## Examples
 
 The examples below are "raw" ASGI applications with minimal configurations. You are more likely than not going to be using a framework, but you should be able to replace the `app` in these example with most ASGI framework applications. Please read the [HTTP](https://erm.github.io/mangum/http/) and [WebSocket](https://erm.github.io/mangum/websocket/) docs for more detailed configuration information.
@@ -163,11 +212,6 @@ async def app(scope, receive, send):
 
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "s3",
-        "params": {
-            "bucket": "<s3-bucket-to-store-connections>"
-        }
-    }
+    dsn="s3://my-bucket-12345"
 )
 ```
diff --git a/docs/http.md b/docs/http.md
index 18905438..828d6eca 100644
--- a/docs/http.md
+++ b/docs/http.md
@@ -2,36 +2,12 @@
 
 Mangum provides support for [HTTP](https://docs.aws.amazon.com/apigateway/latest/developerguide/http-api.html) and [REST](https://docs.aws.amazon.com/apigateway/latest/developerguide/apigateway-rest-api.html) APIs in API Gateway. The adapter class handles parsing the incoming requests and managing the ASGI cycle.
 
-## Configuration
+## Usage
 
 ```python
-handler = Mangum(
-    app,
-    enable_lifespan=True,
-    log_level="info",
-    api_gateway_base_path=None,
-    text_mime_types=None,
-)
+handler = Mangum(app)
 ```
 
-The adapter class accepts the following optional arguments:
-
-- `enable_lifespan` : **bool** (default=`True`)
-    
-    Specify whether or not to enable lifespan support. The adapter will automatically determine if lifespan is supported by the framework unless explicitly disabled.
-
-- `log_level` : **str** (default=`"info"`)
-    
-    Level parameter for the logger.
-
-- `api_gateway_base_path` : **str**
-    
-    Base path to strip from URL when using a custom domain name.
-
-- `text_mime_types` : **list**
-        
-    The list of MIME types (in addition to the defaults) that should not return binary responses in API Gateway.
-
 ## Binary support
 
 Binary response support is available depending on the `Content-Type` and `Content-Encoding` headers. The default text mime types are the following:
diff --git a/docs/introduction.md b/docs/introduction.md
index eb64856b..48c6b5d2 100644
--- a/docs/introduction.md
+++ b/docs/introduction.md
@@ -48,7 +48,7 @@ pip install mangum[redis]==0.9.0b1
 
 ## Usage
 
-The `Mangum` adapter class is designed to wrap any ASGI application, accepting various configuration options, returning a callable. It can wrap an application and be assigned to the handler:
+The `Mangum` adapter class is designed to wrap any ASGI application and returns a callable. It can wrap an application and be assigned to the handler:
 
 ```python
 from mangum import Mangum
@@ -71,6 +71,55 @@ def handler(event, context):
     return response
 ```
 
+## Configuration
+
+The adapter accepts various arguments for configuring lifespan, logging, HTTP, WebSocket, and API Gateway behaviour.
+
+### Usage
+
+```python
+handler = Mangum(
+    app,
+    enable_lifespan=True,
+    log_level="info",
+    api_gateway_base_path=None,
+    text_mime_types=None,
+    dsn=None,
+    api_gateway_endpoint_url=None,
+    api_gateway_region_name=None
+)
+```
+
+#### Parameters
+
+- `enable_lifespan` : **bool**
+    
+    Specify whether or not to enable lifespan support. The adapter will automatically determine if lifespan is supported by the framework unless explicitly disabled.
+
+- `log_level` : **str**
+    
+    Level parameter for the logger.
+
+- `api_gateway_base_path` : **str**
+    
+    Base path to strip from URL when using a custom domain name.
+
+- `text_mime_types` : **list**
+        
+    The list of MIME types (in addition to the defaults) that should not return binary responses in API Gateway.
+
+- `dsn`: **str*
+    
+    DSN connection string to configure a supported WebSocket backend.
+
+- `api_gateway_endpoint_url` : **str**
+    
+    The endpoint url to use when sending data to WebSocket connections in API Gateway. This is useful if you are debugging locally with a package such as [serverless-dynamodb-local](https://github.com/99xt/serverless-dynamodb-local).
+
+- `api_gateway_region_name` : **str**
+    
+    The region name of the API Gateway that is managing the API connections.
+
 ## Examples
 
 The examples below are "raw" ASGI applications with minimal configurations. You are more likely than not going to be using a framework, but you should be able to replace the `app` in these example with most ASGI framework applications. Please read the [HTTP](https://erm.github.io/mangum/http/) and [WebSocket](https://erm.github.io/mangum/websocket/) docs for more detailed configuration information.
@@ -163,11 +212,6 @@ async def app(scope, receive, send):
 
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "s3",
-        "params": {
-            "bucket": "<s3-bucket-to-store-connections>"
-        }
-    }
+    dsn="s3://my-bucket-12345"
 )
 ```
diff --git a/docs/websockets.md b/docs/websockets.md
index 891da677..43397120 100644
--- a/docs/websockets.md
+++ b/docs/websockets.md
@@ -18,122 +18,83 @@ A connected client has sent a message. The adapter will retrieve the initial req
 
 The client or the server disconnects from the API. The adapter will remove the connection from the backend.
 
-## Backends
+### Backends
 
 A data source, such as a cloud database, is required in order to persist the connection identifiers in a 'serverless' environment. Any data source can be used as long as it is accessible remotely to the AWS Lambda function.
 
-All supported backends require a `params` configuration mapping. The `ws_config` configuration must contain the name of a backend along with any required `params` for the selected backend.
-
-#### Configuration
+All supported backends require a `dsn` connection string argument to configure the connection. Backends that already support connection strings, such as PostgreSQL and Redis, can use their existing syntax. Other backends, such as S3 and DynamoDB, are parsed with a custom syntax defined in the backend class.
 
 ```python
-handler = Mangum(
-    app,
-    ws_config={
-        "backend": "postgresql|redis|dynamodb|s3|sqlite3",
-        "params": {...},
-    },
-)
+handler = Mangum(app, dsn="[postgresql|redis|dynamodb|s3|sqlite]://[...]")
 ```
 
-##### Required
-
-- `ws_config` : **dict** *(required)*
-
-    Configuration mapping for a supported WebSocket backend.
-
-The following required values need to be defined inside the `ws_config`:
-
-- `backend` : **str** *(required)*
-
-    Name of data source backend to use. 
-
-
-- `params` : **dict** *(required)*
-    
-    Parameter mapping of required and optional arguments for the specified backend.
-
 The following backends are currently supported:
 
  - `dynamodb`
  - `s3`
  - `postgresql`
  - `redis`
- - `sqlite3` (for local debugging)
-
-##### Optional
-
-The following optional values may be defined inside the `ws_config`:
-
-- `api_gateway_endpoint_url` : **str**
-    
-    The endpoint url to use in API Gateway Management API calls. This is useful if you are debugging locally with a package such as [serverless-offline](https://github.com/dherault/serverless-offline).
-
-- `api_gateway_region_name` : **str**
-    
-    The region name of the API Gateway that is managing the API connections.
+ - `sqlite` (for local debugging)
 
 ### DynamoDB
 
-The `DynamoDBackend` uses a [DynamoDB](https://aws.amazon.com/dynamodb/) table to store the connection details.
+The `DynamoDBBackend` uses a [DynamoDB](https://aws.amazon.com/dynamodb/) table to store the connection details.
 
-#### Configuration
+#### Usage
 
 ```python
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "dynamodb",
-        "params":{
-            "table_name": "connections"
-        },
-    },
+    dsn="dynamodb://mytable"
 )
 ```
 
-##### Required
+##### Parameters
+
+The DynamoDB backend `dsn` uses the following connection string syntax:
 
-- `table_name` : **str** *(required)*
+```
+dynamodb://<table_name>[?region=<region-name>&endpoint_url=<url>]
+```
 
-    The name of the table in DynamoDB.
+- `table_name` (Required)
 
-##### Optional
+    The name of the table in DynamoDB.  
 
-- `region_name` : **str**
+- `region_name`
     
     The region name of the DyanmoDB table.
 
-- `endpoint_url`: **str**
+- `endpoint_url`
 
     The endpoint url to use in DynamoDB calls. This is useful if you are debugging locally with a package such as [serverless-dynamodb-local](https://github.com/99xt/serverless-dynamodb-local).
 
 ### S3
 
-The `S3Backend` uses an [Amazon S3](https://aws.amazon.com/s3/](https://aws.amazon.com/s3/) bucket as a key-value store to store the connection details.
+The `S3Backend` uses an (S3)[https://aws.amazon.com/s3/](https://aws.amazon.com/s3/)] bucket as a key-value store to store the connection details.
 
-#### Configuration
+#### Usage
 
 ```python
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "s3",
-        "params": {
-            "bucket": "asgi-websocket-connections-12345"
-        },
-    },
+    dsn="s3://my-bucket-12345"
 )
 ```
 
-##### Required
+##### Parameters
+
+The S3 backend `dsn` uses the following connection string syntax:
+
+```
+s3://<bucket>[/key/...][?region=<region-name>]
+```
 
-- `bucket` : **str** *(required)*
+- `bucket` (Required)
     
     The name of the bucket in S3.
 
-##### Optional
-
-- `region_name` : **str**
+- `region_name`
     
     The region name of the S3 bucket.
 
@@ -141,126 +102,80 @@ handler = Mangum(
 
 The `PostgreSQLBackend` requires [psycopg2](https://github.com/psycopg/psycopg2) and access to a remote PostgreSQL database.
 
-#### Configuration
+### Usage
 
 ```python
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "postgresql",
-        "params": {
-            "database": "mangum",
-            "user": "postgres",
-            "password": "correct horse battery staple",
-            "host": "mydb.12345678910.ap-southeast-1.rds.amazonaws.com"
-        },
-    },
+    dsn="postgresql://myuser:mysecret@my.host:5432/mydb"
 )
 ```
 
-##### Required
-
-- `uri`: **str** *(required)*
-    The connection string for the remote database.
-
-If a `uri` is not supplied, then the following parameters are required:
-
-- `database` : **str** *(required)*
-    
-    The name of the database.
-
-- `user` : **str** *(required)*
-    
-    Postgres user username.
+##### Parameters
 
-- `password` : **str** *(required)*
-    
-    Postgres user password.
-
-- `host` : **str** *(required)*
-    
-    Host for Postgres database connection.
+The PostgreSQL backend `dsn` uses the following connection string syntax:
 
-##### Optional
+```
+postgresql://[user[:password]@][host][:port][,...][/dbname][?param1=value1&...]
+```
 
-- `port` : **str** (default=`5432`)
-    
-    Port number for Postgres database connection.
+`host` (Required)
 
-- `connect_timeout` **int** (default=`5`)
-    
-    Timeout for database connection.
+    The network location of the PostgreSQL database
 
-- `table_name` **str (default=`"connection"`)
-    
-    Table name to use to store WebSocket connections. 
+Read more about the supported uri schemes and additional parameters [here](https://www.postgresql.org/docs/10/libpq-connect.html#LIBPQ-CONNSTRING).
 
 ### Redis
 
-The `RedisBackend` requires [redis-py](https://github.com/andymccurdy/redis-py) and access to a Redis server.
+The `RedisBackend` requires (redis-py)[https://github.com/andymccurdy/redis-py] and access to a Redis server.
 
-#### Configuration
+### Usage
 
 ```python
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "redis",
-        "params": {
-            "host": "my.redis.host",
-            "port": 6379
-            "password": "correct horse battery staple",
-        },
-    },
+    dsn="redis://:mysecret@my.host:6379/0"
 )
 ```
 
-##### Required
+#### Parameters
 
-- `host` : **str** *(required)*
-    
-    Host for Redis server.
+The Redis backend `dsn` uses the following connection string syntax:
 
-##### Optional
+```
+redis://[[user:]password@]host[:port][/database]
+```
 
-- `port` : **str** (default=`6379`)
+- `host` (required)
     
-    Port number for Redis server.
+    The network location of the Redis server.
 
-- `password` : **str**
-    
-    Password for Redis server.
+Read more about the supported uri schemes and additional parameters [here](https://www.iana.org/assignments/uri-schemes/prov/redis).
 
-### SQLite
+### SQlite3
 
-The `sqlite3` backend uses a local [sqlite3](https://docs.python.org/3/library/sqlite3.html) database to store connection. It is intended for ***local*** debugging (with a package such as [Serverless Offline](https://github.com/dherault/serverless-offline)) and will ***not*** work in an AWS Lambda deployment.
+The `sqlite` backend uses a local [sqlite3](https://docs.python.org/3/library/sqlite3.html) database to store connection. It is intended for ***local*** debugging (with a package such as [Serverless Offline](https://github.com/dherault/serverless-offline)) and will ***not*** work in an AWS Lambda deployment.
 
-#### Configuration
+### Usage
 
 ```python
 handler = Mangum(
     app,
-    ws_config={
-        "backend": "sqlite3",
-        "params": {
-            "file_path": "mangum.sqlite3",
-            "table_name": "connection",
-        },
-    },
+    dsn="sqlite://mydbfile.sqlite3"
 )
 ```
 
-##### Required
+#### Parameters
 
-- `file_path` : **str** *(required)*
+The SQLite backend uses the following connection string syntax:
 
-    The file name or path to a sqlite3 file. If one does not exist, then it will be created automatically.
-
-##### Optional
+```
+sqlite://[file_path].db
+```
 
-- `table_name` : **str** (default=`"connection"`)
+- `file_path` (Requred)
 
-    The name of the table to use for the connections in an sqlite3 database.
+    The file name or path to an sqlite3 database file. If one does not exist, then it will be created automatically.
 
 ### Alternative backends
 
diff --git a/mangum/adapter.py b/mangum/adapter.py
index 841121fb..c99d8557 100644
--- a/mangum/adapter.py
+++ b/mangum/adapter.py
@@ -58,7 +58,9 @@ class Mangum:
     log_level: str = "info"
     api_gateway_base_path: typing.Optional[str] = None
     text_mime_types: typing.Optional[typing.List[str]] = None
-    ws_config: typing.Optional[dict] = None
+    dsn: typing.Optional[str] = None
+    api_gateway_endpoint_url: typing.Optional[str] = None
+    api_gateway_region_name: typing.Optional[str] = None
 
     def __post_init__(self) -> None:
         self.logger = get_logger(self.log_level)
@@ -163,33 +165,32 @@ def handle_http(self, event: dict, context: dict, *, is_http_api: bool) -> dict:
         return response
 
     def handle_ws(self, event: dict, context: dict) -> dict:
-        if self.ws_config is None:
+        if self.dsn is None:
             raise ConfigurationError(
-                "A `ws_config` configuration mapping is required for WebSocket support."
+                "A `dsn` connection string is required for WebSocket support."
             )
 
         event_type = event["requestContext"]["eventType"]
         connection_id = event["requestContext"]["connectionId"]
         stage = event["requestContext"]["stage"]
         domain_name = event["requestContext"]["domainName"]
+        self.logger.info(
+            "%s event received for WebSocket connection %s", event_type, connection_id
+        )
 
-        api_gateway_endpoint_url = self.ws_config.get(
-            "api_gateway_endpoint_url", f"https://{domain_name}/{stage}"
+        api_gateway_endpoint_url = (
+            self.api_gateway_endpoint_url or f"https://{domain_name}/{stage}"
         )
-        api_gateway_region_name = self.ws_config.get(
-            "api_gateway_region_name", os.environ["AWS_REGION"]
+        api_gateway_region_name = (
+            self.api_gateway_region_name or os.environ["AWS_REGION"]
         )
-
         websocket = WebSocket(
             connection_id,
-            ws_config=self.ws_config,
+            dsn=self.dsn,
             api_gateway_endpoint_url=api_gateway_endpoint_url,
             api_gateway_region_name=api_gateway_region_name,
         )
 
-        self.logger.info(
-            "%s event received for WebSocket connection %s", event_type, connection_id
-        )
         if event_type == "CONNECT":
 
             headers = (
diff --git a/mangum/backends/base.py b/mangum/backends/base.py
index 30b5f1a5..4a450bb0 100644
--- a/mangum/backends/base.py
+++ b/mangum/backends/base.py
@@ -4,10 +4,13 @@
 @dataclass
 class WebSocketBackend:
     """
-    Base class for implementing WebSocket backends to store APIGateway connections.
+    Base class for implementing WebSocket backends to store API Gateway connections.
+
+    WebSocket backends are required to implement configuration based on a `dsn`
+    connection string.
     """
 
-    params: dict
+    dsn: str
 
     def create(self, connection_id: str, initial_scope: str) -> None:
         """
diff --git a/mangum/backends/dynamodb.py b/mangum/backends/dynamodb.py
index 5611d461..dc8a918c 100644
--- a/mangum/backends/dynamodb.py
+++ b/mangum/backends/dynamodb.py
@@ -1,30 +1,41 @@
 import os
+
+from urllib.parse import urlparse, parse_qs
 from dataclasses import dataclass
 
 import boto3
-from botocore.exceptions import ClientError
+from botocore.config import Config
+from botocore.exceptions import ClientError, EndpointConnectionError
 
 from mangum.backends.base import WebSocketBackend
-from mangum.exceptions import WebSocketError, ConfigurationError
+from mangum.exceptions import WebSocketError
 
 
 @dataclass
 class DynamoDBBackend(WebSocketBackend):
     def __post_init__(self) -> None:
-        try:
-            table_name = self.params["table_name"]
-        except KeyError:
-            raise ConfigurationError("DynamoDB 'table_name' missing.")
-        region_name = self.params.get("region_name", os.environ["AWS_REGION"])
-        endpoint_url = self.params.get("endpoint_url", None)
+        parsed_dsn = urlparse(self.dsn)
+        parsed_query = parse_qs(parsed_dsn.query)
+        table_name = parsed_dsn.hostname
+
+        region_name = (
+            parsed_query["region"][0]
+            if "region" in parsed_query
+            else os.environ["AWS_REGION"]
+        )
+        endpoint_url = (
+            parsed_query["endpoint_url"][0] if "endpoint_url" in parsed_query else None
+        )
         try:
             dynamodb_resource = boto3.resource(
-                "dynamodb", region_name=region_name, endpoint_url=endpoint_url
+                "dynamodb",
+                region_name=region_name,
+                endpoint_url=endpoint_url,
+                config=Config(connect_timeout=2, retries={"max_attempts": 0}),
             )
             dynamodb_resource.meta.client.describe_table(TableName=table_name)
-        except ClientError as exc:
+        except (EndpointConnectionError, ClientError) as exc:
             raise WebSocketError(exc)
-
         self.connection = dynamodb_resource.Table(table_name)
 
     def create(self, connection_id: str, initial_scope: str) -> None:
diff --git a/mangum/backends/postgresql.py b/mangum/backends/postgresql.py
index 90a9d7fa..1a008233 100644
--- a/mangum/backends/postgresql.py
+++ b/mangum/backends/postgresql.py
@@ -1,87 +1,43 @@
-import logging
 from dataclasses import dataclass
 
 import psycopg2
-from psycopg2 import sql
 
 from mangum.backends.base import WebSocketBackend
-from mangum.exceptions import ConfigurationError
 
 
 @dataclass
 class PostgreSQLBackend(WebSocketBackend):
     def __post_init__(self) -> None:
-        self.logger = logging.getLogger("mangum.websocket.postgres")
-        self.logger.debug("Connecting to PostgreSQL database.")
-        connect_timeout = self.params.get("connect_timeout", 5)
-        if "uri" in self.params:
-            self.connection = psycopg2.connect(
-                self.params["uri"], connect_timeout=connect_timeout
-            )
-        else:
-            try:
-                database = self.params["database"]
-                user = self.params["user"]
-                password = self.params["password"]
-                host = self.params["host"]
-            except KeyError:  # pragma: no cover
-                raise ConfigurationError("PostgreSQL connection details missing.")
-            port = self.params.get("port", "5432")  # pragma: no cover
-            self.connection = psycopg2.connect(
-                database=database,
-                user=user,
-                password=password,
-                host=host,
-                port=port,
-                connect_timeout=connect_timeout,
-            )
-        self.table_name = self.params.get("table_name", "mangum")
+        self.connection = psycopg2.connect(self.dsn, connect_timeout=5)
         self.cursor = self.connection.cursor()
         self.cursor.execute(
-            sql.SQL(
-                "create table if not exists {} (id varchar(64) primary key, initial_scope text)"
-            ).format(sql.Identifier(self.table_name))
+            "create table if not exists mangum_websockets (id varchar(64) primary key, initial_scope text)"
         )
         self.connection.commit()
-        self.logger.debug("Connection established.")
 
     def create(self, connection_id: str, initial_scope: str) -> None:
-        self.logger.debug("Creating database entry for %s", connection_id)
         self.cursor.execute(
-            sql.SQL("insert into {} values (%s, %s)").format(
-                sql.Identifier(self.table_name)
-            ),
+            "insert into mangum_websockets values (%s, %s)",
             (connection_id, initial_scope),
         )
-
         self.connection.commit()
         self.connection.close()
-        self.logger.debug("Database entry created.")
 
     def fetch(self, connection_id: str) -> str:
-        self.logger.debug("Fetching initial scope for %s", connection_id)
         self.cursor.execute(
-            sql.SQL("select initial_scope from {} where id = %s").format(
-                sql.Identifier(self.table_name)
-            ),
+            "select initial_scope from mangum_websockets where id = %s",
             (connection_id,),
         )
         initial_scope = self.cursor.fetchone()[0]
         self.cursor.close()
         self.connection.close()
-        self.logger.debug("Initial scope fetched.")
 
         return initial_scope
 
     def delete(self, connection_id: str) -> None:
-        self.logger.debug("Deleting database entry for %s", connection_id)
         self.cursor.execute(
-            sql.SQL("delete from {} where id = %s").format(
-                sql.Identifier(self.table_name)
-            ),
-            (connection_id,),
+            "delete from mangum_websockets where id = %s", (connection_id,)
         )
         self.connection.commit()
         self.cursor.close()
         self.connection.close()
-        self.logger.debug("Database entry deleted.")
diff --git a/mangum/backends/redis.py b/mangum/backends/redis.py
index 270c8175..0b4a8b96 100644
--- a/mangum/backends/redis.py
+++ b/mangum/backends/redis.py
@@ -1,39 +1,22 @@
-import logging
 from dataclasses import dataclass
 
 import redis
 
 from mangum.backends.base import WebSocketBackend
-from mangum.exceptions import ConfigurationError
 
 
 @dataclass
 class RedisBackend(WebSocketBackend):
     def __post_init__(self) -> None:
-        self.logger = logging.getLogger("mangum.websocket.redis")
-        self.logger.debug("Connecting to Redis host.")
-        try:
-            host = self.params["host"]
-        except KeyError:  # pragma: no cover
-            raise ConfigurationError("Redis 'host' parameter missing.")
-        password = self.params.get("password")
-        port = self.params.get("port", "5432")  # pragma: no cover
-        self.connection = redis.Redis(host, port, password=password)
-        self.logger.debug("Connection established.")
+        self.connection = redis.Redis.from_url(self.dsn)
 
     def create(self, connection_id: str, initial_scope: str) -> None:
-        self.logger.debug("Creating entry for %s", connection_id)
         self.connection.set(connection_id, initial_scope)
-        self.logger.debug("Entry created.")
 
     def fetch(self, connection_id: str) -> str:
-        self.logger.debug("Fetching initial scope for %s", connection_id)
         initial_scope = self.connection.get(connection_id)
-        self.logger.debug("Initial scope fetched.")
 
         return initial_scope
 
     def delete(self, connection_id: str) -> None:
-        self.logger.debug("Deleting entry for %s", connection_id)
         self.connection.delete(connection_id)
-        self.logger.debug("Entry deleted.")
diff --git a/mangum/backends/s3.py b/mangum/backends/s3.py
index 04763e26..540424d3 100644
--- a/mangum/backends/s3.py
+++ b/mangum/backends/s3.py
@@ -1,40 +1,48 @@
 import os
 import logging
 from dataclasses import dataclass
+from urllib.parse import urlparse, parse_qs
+
 
 import boto3
 from botocore.client import Config
 from botocore.exceptions import ClientError
 
 from mangum.backends.base import WebSocketBackend
-from mangum.exceptions import ConfigurationError
+
+
+logger = logging.getLogger("mangum.websocket.s3")
 
 
 @dataclass
 class S3Backend(WebSocketBackend):
     def __post_init__(self) -> None:
-        self.logger = logging.getLogger("mangum.websocket.s3")
-        try:
-            self.bucket = self.params["bucket"]
-        except KeyError:
-            raise ConfigurationError("S3 'bucket' parameter missing.")
-        region_name = self.params.get("region_name", os.environ["AWS_REGION"])
+        parsed_dsn = urlparse(self.dsn)
+        parsed_query = parse_qs(parsed_dsn.query)
+        self.bucket = parsed_dsn.hostname
+        if parsed_dsn.path and parsed_dsn.path != "/":
+            if not parsed_dsn.path.endswith("/"):
+                self.key = parsed_dsn.path + "/"
+            else:
+                self.key = parsed_dsn.path
+        else:
+            self.key = ""
+        region_name = parsed_query.get("region", os.environ["AWS_REGION"])
+        create_bucket = False
         self.connection = boto3.client(
             "s3",
             region_name=region_name,
-            config=Config(connect_timeout=5, retries={"max_attempts": 0}),
+            config=Config(connect_timeout=2, retries={"max_attempts": 0}),
         )
-        create_bucket = False
         try:
             self.connection.head_bucket(Bucket=self.bucket)
         except ClientError as exc:
             error_code = int(exc.response["Error"]["Code"])
             if error_code == 403:  # pragma: no cover
-                self.logger.error("S3 bucket access forbidden!")
+                logger.error("S3 bucket access forbidden!")
             elif error_code == 404:
-                self.logger.info(f"Bucket {self.bucket} not found, creating.")
+                logger.info(f"Bucket {self.bucket} not found, creating.")
                 create_bucket = True
-
         if create_bucket:
             self.connection.create_bucket(
                 Bucket=self.bucket,
@@ -43,14 +51,20 @@ def __post_init__(self) -> None:
 
     def create(self, connection_id: str, initial_scope: str) -> None:
         self.connection.put_object(
-            Body=initial_scope.encode(), Bucket=self.bucket, Key=connection_id
+            Body=initial_scope.encode(),
+            Bucket=self.bucket,
+            Key=f"{self.key}{connection_id}",
         )
 
     def fetch(self, connection_id: str) -> str:
-        s3_object = self.connection.get_object(Bucket=self.bucket, Key=connection_id)
+        s3_object = self.connection.get_object(
+            Bucket=self.bucket, Key=f"{self.key}{connection_id}"
+        )
         initial_scope = s3_object["Body"].read().decode()
 
         return initial_scope
 
     def delete(self, connection_id: str) -> None:
-        self.connection.delete_object(Bucket=self.bucket, Key=connection_id)
+        self.connection.delete_object(
+            Bucket=self.bucket, Key=f"{self.key}{connection_id}"
+        )
diff --git a/mangum/backends/sqlite.py b/mangum/backends/sqlite.py
new file mode 100644
index 00000000..9d6f65bf
--- /dev/null
+++ b/mangum/backends/sqlite.py
@@ -0,0 +1,41 @@
+import sqlite3
+
+from dataclasses import dataclass
+
+from mangum.backends.base import WebSocketBackend
+
+
+@dataclass
+class SQLiteBackend(WebSocketBackend):
+    def __post_init__(self) -> None:
+        dsn = self.dsn.replace("sqlite://", "")
+        self.connection = sqlite3.connect(dsn)
+        self.cursor = self.connection.cursor()
+        self.cursor.execute(
+            f"create table if not exists mangum_ws (id varchar(64) primary key, initial_scope text)"
+        )
+
+    def create(self, connection_id: str, initial_scope: str) -> None:
+        self.cursor.execute(
+            f"insert into mangum_ws values (?, ?)", (connection_id, initial_scope)
+        )
+        self.connection.commit()
+        self.cursor.close()
+        self.connection.close()
+
+    def fetch(self, connection_id: str) -> str:
+        initial_scope = self.cursor.execute(
+            f"select initial_scope from mangum_ws where id = ?", (connection_id,)
+        ).fetchone()[0]
+        self.cursor.close()
+        self.connection.close()
+
+        return initial_scope
+
+    def delete(self, connection_id: str) -> None:
+        self.cursor.execute(
+            f"delete from mangum_ws where id = ?", (connection_id,)
+        ).fetchone()
+        self.connection.commit()
+        self.cursor.close()
+        self.connection.close()
diff --git a/mangum/backends/sqlite3.py b/mangum/backends/sqlite3.py
deleted file mode 100644
index b3cfbad1..00000000
--- a/mangum/backends/sqlite3.py
+++ /dev/null
@@ -1,53 +0,0 @@
-import os
-import sqlite3
-
-from dataclasses import dataclass
-
-from mangum.backends.base import WebSocketBackend
-from mangum.exceptions import ConfigurationError
-
-
-@dataclass
-class SQLiteBackend(WebSocketBackend):
-    def __post_init__(self) -> None:
-        try:
-            file_path = self.params["file_path"]
-        except KeyError:
-            raise ConfigurationError(f"SQLite3 database 'file_path' missing.")
-        self.table_name = self.params.get("table_name", "mangum")
-        if not os.path.exists(file_path):
-            self.connection = sqlite3.connect(file_path)
-            self.cursor = self.connection.cursor()
-            self.cursor.execute(
-                f"create table {self.table_name} (id varchar(64) primary key, initial_scope text)"
-            )
-        else:
-            self.connection = sqlite3.connect(file_path)
-            self.cursor = self.connection.cursor()
-
-    def create(self, connection_id: str, initial_scope: str) -> None:
-        self.cursor.execute(
-            f"insert into {self.table_name} values (?, ?)",
-            (connection_id, initial_scope),
-        )
-        self.connection.commit()
-        self.cursor.close()
-        self.connection.close()
-
-    def fetch(self, connection_id: str) -> str:
-        initial_scope = self.cursor.execute(
-            f"select initial_scope from {self.table_name} where id = ?",
-            (connection_id,),
-        ).fetchone()[0]
-        self.cursor.close()
-        self.connection.close()
-
-        return initial_scope
-
-    def delete(self, connection_id: str) -> None:
-        self.cursor.execute(
-            f"delete from {self.table_name} where id = ?", (connection_id,)
-        ).fetchone()
-        self.connection.commit()
-        self.cursor.close()
-        self.connection.close()
diff --git a/mangum/websocket.py b/mangum/websocket.py
index e0ab5516..291e54d7 100644
--- a/mangum/websocket.py
+++ b/mangum/websocket.py
@@ -1,6 +1,7 @@
 import json
 import logging
 from dataclasses import dataclass
+from urllib.parse import urlparse
 
 from mangum.types import Scope
 from mangum.exceptions import WebSocketError, ConfigurationError
@@ -14,52 +15,58 @@
 class WebSocket:
 
     connection_id: str
-    ws_config: dict
+    dsn: str
     api_gateway_region_name: str
     api_gateway_endpoint_url: str
 
     def __post_init__(self) -> None:
         self.logger: logging.Logger = logging.getLogger("mangum.websocket")
-        try:
-            backend = self.ws_config["backend"]
-            params = self.ws_config["params"]
-        except KeyError as exc:
-            raise ConfigurationError(f"WebSocket config {exc} missing.")
-        if backend == "sqlite3":
+        parsed_dsn = urlparse(self.dsn)
+        if not any((parsed_dsn.hostname, parsed_dsn.path)):
+            raise ConfigurationError("Invalid value for `dsn` provided.")
+        scheme = parsed_dsn.scheme
+        self.logger.debug(
+            f"Attempting WebSocket backend connection using scheme: {scheme}"
+        )
+        if scheme == "sqlite":
             self.logger.info(
                 "The `SQLiteBackend` should be only be used for local "
                 "debugging. It will not work in a deployed environment."
             )
-            from mangum.backends.sqlite3 import SQLiteBackend
+            from mangum.backends.sqlite import SQLiteBackend
 
-            self._backend = SQLiteBackend(params)  # type: ignore
-        elif backend == "dynamodb":
+            self._backend = SQLiteBackend(self.dsn)  # type: ignore
+        elif scheme == "dynamodb":
             from mangum.backends.dynamodb import DynamoDBBackend
 
-            self._backend = DynamoDBBackend(params)  # type: ignore
-        elif backend == "s3":
+            self._backend = DynamoDBBackend(self.dsn)  # type: ignore
+        elif scheme == "s3":
             from mangum.backends.s3 import S3Backend
 
-            self._backend = S3Backend(params)  # type: ignore
+            self._backend = S3Backend(self.dsn)  # type: ignore
 
-        elif backend == "postgresql":
+        elif scheme == "postgresql":
             from mangum.backends.postgresql import PostgreSQLBackend
 
-            self._backend = PostgreSQLBackend(params)  # type: ignore
+            self._backend = PostgreSQLBackend(self.dsn)  # type: ignore
 
-        elif backend == "redis":
+        elif scheme == "redis":
             from mangum.backends.redis import RedisBackend
 
-            self._backend = RedisBackend(params)  # type: ignore
+            self._backend = RedisBackend(self.dsn)  # type: ignore
 
         else:
-            raise ConfigurationError(f"{backend} is not a supported backend.")
+            raise ConfigurationError(f"{scheme} does not match a supported backend.")
+        self.logger.debug("WebSocket backend connection established.")
 
     def create(self, initial_scope: dict) -> None:
+        self.logger.debug("Creating scope entry for %s", self.connection_id)
         initial_scope_json = json.dumps(initial_scope)
         self._backend.create(self.connection_id, initial_scope_json)
+        self.logger.debug("Scope entry created.")
 
     def fetch(self) -> None:
+        self.logger.debug("Fetching scope entry for %s", self.connection_id)
         initial_scope = self._backend.fetch(self.connection_id)
         scope = json.loads(initial_scope)
         query_string = scope["query_string"]
@@ -68,9 +75,12 @@ def fetch(self) -> None:
             headers = [[k.encode(), v.encode()] for k, v in headers.items() if headers]
         scope.update({"headers": headers, "query_string": query_string.encode()})
         self.scope: Scope = scope
+        self.logger.debug("Scope entry fetched.")
 
     def delete(self) -> None:
+        self.logger.debug("Deleting scope entry for %s", self.connection_id)
         self._backend.delete(self.connection_id)
+        self.logger.debug("Scope entry deleted.")
 
     def post_to_connection(self, msg_data: bytes) -> None:  # pragma: no cover
         try:
diff --git a/tests/conftest.py b/tests/conftest.py
index 21dcdbdb..b2c1d33d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -286,5 +286,16 @@ def mock_ws_disconnect_event() -> dict:
     }
 
 
+@pytest.fixture
+def mock_websocket_app():
+    async def app(scope, receive, send):
+        await send({"type": "websocket.accept", "subprotocol": None})
+        await send({"type": "websocket.send", "text": "Hello world!"})
+        await send({"type": "websocket.send", "bytes": b"Hello world!"})
+        await send({"type": "websocket.close", "code": 1000})
+
+    return app
+
+
 def pytest_generate_tests(metafunc):
     os.environ["AWS_REGION"] = "ap-southeast-1"
diff --git a/tests/test_backends.py b/tests/test_backends.py
deleted file mode 100644
index 53afef56..00000000
--- a/tests/test_backends.py
+++ /dev/null
@@ -1,217 +0,0 @@
-import os
-import mock
-
-import pytest
-import boto3
-import redis
-import testing.redis
-import testing.postgresql
-from sqlalchemy import create_engine
-from moto import mock_dynamodb2, mock_s3
-
-from mangum import Mangum
-from mangum.exceptions import WebSocketError, ConfigurationError
-
-
-def test_sqlite_3_backend(
-    tmp_path, mock_ws_connect_event, mock_ws_send_event, mock_ws_disconnect_event
-) -> None:
-    valid = {
-        "backend": "sqlite3",
-        "params": {"file_path": os.path.join(tmp_path, "db.sqlite3")},
-    }
-
-    missing_file_path = {"backend": "sqlite3", "params": {}}
-
-    async def app(scope, receive, send):
-        await send({"type": "websocket.accept", "subprotocol": None})
-        await send({"type": "websocket.send", "text": "Hello world!"})
-        await send({"type": "websocket.send", "bytes": b"Hello world!"})
-        await send({"type": "websocket.close", "code": 1000})
-
-    handler = Mangum(app, ws_config=valid)
-    response = handler(mock_ws_connect_event, {})
-    assert response == {"statusCode": 200}
-
-    handler = Mangum(app, ws_config=valid)
-    with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
-        send.return_value = None
-        response = handler(mock_ws_send_event, {})
-        assert response == {"statusCode": 200}
-
-    handler = Mangum(app, ws_config=valid)
-    response = handler(mock_ws_disconnect_event, {})
-    assert response == {"statusCode": 200}
-
-    # Missing file path
-    with pytest.raises(ConfigurationError):
-        handler = Mangum(app, ws_config=missing_file_path)
-        response = handler(mock_ws_connect_event, {})
-
-
-@mock_dynamodb2
-def test_dynamodb_backend(
-    mock_ws_connect_event, mock_ws_send_event, mock_ws_disconnect_event
-) -> None:
-    table_name = "mangum"
-    region_name = "ap-southeast-1"
-    dynamodb_resource = boto3.resource("dynamodb", region_name=region_name)
-    dynamodb_resource.meta.client.create_table(
-        TableName=table_name,
-        KeySchema=[{"AttributeName": "connectionId", "KeyType": "HASH"}],
-        AttributeDefinitions=[{"AttributeName": "connectionId", "AttributeType": "S"}],
-        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
-    )
-    table = dynamodb_resource.Table(table_name)
-
-    valid = {
-        "backend": "dynamodb",
-        "params": {"table_name": table_name, "region_name": region_name},
-    }
-    table_does_not_exist = {
-        "backend": "dynamodb",
-        "params": {"table_name": "does-not-exist", "region_name": region_name},
-    }
-    missing_table_name = {"backend": "dynamodb", "params": {"region_name": region_name}}
-
-    async def app(scope, receive, send):
-        await send({"type": "websocket.accept", "subprotocol": None})
-        await send({"type": "websocket.send", "text": "Hello world!"})
-        await send({"type": "websocket.send", "bytes": b"Hello world!"})
-        await send({"type": "websocket.close", "code": 1000})
-
-    # Test valid cases
-    handler = Mangum(app, ws_config=valid)
-    response = handler(mock_ws_connect_event, {})
-    assert response == {"statusCode": 200}
-
-    handler = Mangum(app, ws_config=valid)
-    with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
-        send.return_value = None
-        response = handler(mock_ws_send_event, {})
-        assert response == {"statusCode": 200}
-
-    handler = Mangum(app, ws_config=valid)
-    response = handler(mock_ws_disconnect_event, {})
-    assert response == {"statusCode": 200}
-
-    # Table does not exist
-    handler = Mangum(app, ws_config=table_does_not_exist)
-    with pytest.raises(WebSocketError):
-        response = handler(mock_ws_connect_event, {})
-
-    # Missing table name
-    handler = Mangum(app, ws_config=missing_table_name)
-    with pytest.raises(ConfigurationError):
-        response = handler(mock_ws_connect_event, {})
-
-    # Missing connection
-    table.delete_item(Key={"connectionId": "d4NsecoByQ0CH-Q="})
-    handler = Mangum(app, ws_config=valid)
-    with pytest.raises(WebSocketError):
-        response = handler(mock_ws_send_event, {})
-
-
-@mock_s3
-def test_s3_backend(
-    tmp_path, mock_ws_connect_event, mock_ws_send_event, mock_ws_disconnect_event
-) -> None:
-    bucket = "mangum"
-    conn = boto3.resource("s3")
-    conn.create_bucket(Bucket=bucket)
-    valid = {"backend": "s3", "params": {"bucket": bucket}}
-    missing_bucket = {"backend": "s3", "params": {}}
-    create_new_bucket = {"backend": "s3", "params": {"bucket": "new_bucket"}}
-
-    async def app(scope, receive, send):
-        await send({"type": "websocket.accept", "subprotocol": None})
-        await send({"type": "websocket.send", "text": "Hello world!"})
-        await send({"type": "websocket.send", "bytes": b"Hello world!"})
-        await send({"type": "websocket.close", "code": 1000})
-
-    handler = Mangum(app, ws_config=valid)
-    response = handler(mock_ws_connect_event, {})
-    assert response == {"statusCode": 200}
-
-    handler = Mangum(app, ws_config=valid)
-    with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
-        send.return_value = None
-        response = handler(mock_ws_send_event, {})
-        assert response == {"statusCode": 200}
-
-    handler = Mangum(app, ws_config=valid)
-    response = handler(mock_ws_disconnect_event, {})
-    assert response == {"statusCode": 200}
-
-    # Missing bucket
-    with pytest.raises(ConfigurationError):
-        handler = Mangum(app, ws_config=missing_bucket)
-        response = handler(mock_ws_connect_event, {})
-
-    # Create bucket
-    handler = Mangum(app, ws_config=create_new_bucket)
-    response = handler(mock_ws_connect_event, {})
-    assert response == {"statusCode": 200}
-
-
-def test_postgresql_backend(
-    mock_ws_connect_event, mock_ws_send_event, mock_ws_disconnect_event
-):
-    async def app(scope, receive, send):
-        await send({"type": "websocket.accept", "subprotocol": None})
-        await send({"type": "websocket.send", "text": "Hello world!"})
-        await send({"type": "websocket.send", "bytes": b"Hello world!"})
-        await send({"type": "websocket.close", "code": 1000})
-
-    with testing.postgresql.Postgresql() as postgresql:
-        create_engine(postgresql.url())
-
-        dsn = postgresql.dsn()
-        params = {
-            "uri": f"postgresql://{dsn['user']}:postgres@{dsn['host']}:{dsn['port']}/{dsn['database']}"
-        }
-        handler = Mangum(app, ws_config={"backend": "postgresql", "params": params})
-        response = handler(mock_ws_connect_event, {})
-        assert response == {"statusCode": 200}
-
-        handler = Mangum(app, ws_config={"backend": "postgresql", "params": params})
-        with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
-            send.return_value = None
-            response = handler(mock_ws_send_event, {})
-            assert response == {"statusCode": 200}
-
-        handler = Mangum(app, ws_config={"backend": "postgresql", "params": params})
-        response = handler(mock_ws_disconnect_event, {})
-        assert response == {"statusCode": 200}
-
-        dsn["password"] = "postgres"
-        handler = Mangum(app, ws_config={"backend": "postgresql", "params": dsn})
-        response = handler(mock_ws_connect_event, {})
-        assert response == {"statusCode": 200}
-
-
-def test_redis_backend(
-    mock_ws_connect_event, mock_ws_send_event, mock_ws_disconnect_event
-):
-    async def app(scope, receive, send):
-        await send({"type": "websocket.accept", "subprotocol": None})
-        await send({"type": "websocket.send", "text": "Hello world!"})
-        await send({"type": "websocket.send", "bytes": b"Hello world!"})
-        await send({"type": "websocket.close", "code": 1000})
-
-    with testing.redis.RedisServer() as redis_server:
-        dsn = redis_server.dsn()
-
-        handler = Mangum(app, ws_config={"backend": "redis", "params": dsn})
-        response = handler(mock_ws_connect_event, {})
-        assert response == {"statusCode": 200}
-
-        handler = Mangum(app, ws_config={"backend": "redis", "params": dsn})
-        with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
-            send.return_value = None
-            response = handler(mock_ws_send_event, {})
-            assert response == {"statusCode": 200}
-
-        handler = Mangum(app, ws_config={"backend": "redis", "params": dsn})
-        response = handler(mock_ws_disconnect_event, {})
-        assert response == {"statusCode": 200}
diff --git a/tests/test_websockets.py b/tests/test_websockets.py
index f30d7455..2fe37a44 100644
--- a/tests/test_websockets.py
+++ b/tests/test_websockets.py
@@ -1,46 +1,199 @@
-import pytest
+import mock
 
-import os
+import pytest
+import boto3
+import testing.redis
+import testing.postgresql
+from sqlalchemy import create_engine
+from moto import mock_dynamodb2, mock_s3
 
 from mangum import Mangum
-from mangum.exceptions import ConfigurationError
-
-
-@pytest.mark.parametrize(
-    "ws_config",
-    [
-        {"backend": "unsupported", "params": {}},
-        {"backend": "sqlite3"},
-        {"back": "sqlite3", "params": {}},
-        None,
-    ],
-)
-def test_ws_config(mock_ws_connect_event, ws_config):
-    async def app(scope, receive, send):
-        await send({"type": "websocket.accept", "subprotocol": None})
-        await send({"type": "websocket.send", "text": "Hello world!"})
-        await send({"type": "websocket.send", "bytes": b"Hello world!"})
-        await send({"type": "websocket.close", "code": 1000})
+from mangum.exceptions import WebSocketError, ConfigurationError
 
-    handler = Mangum(app, ws_config=ws_config)
-    with pytest.raises(ConfigurationError):
-        handler(mock_ws_connect_event, {})
 
-
-def test_websocket_cycle_exception(
+def test_websocket_500_error(
     tmp_path, mock_ws_connect_event, mock_ws_send_event
 ) -> None:
     async def app(scope, receive, send):
         await send({"type": "websocket.oops", "subprotocol": None})
 
-    ws_config = {
-        "backend": "sqlite3",
-        "params": {"file_path": os.path.join(tmp_path, "db.sqlite3")},
-    }
+    dsn = f"sqlite://{tmp_path}/mangum.sqlite3"
 
-    handler = Mangum(app, ws_config=ws_config)
+    handler = Mangum(app, dsn=dsn)
     handler(mock_ws_connect_event, {})
 
-    handler = Mangum(app, ws_config=ws_config)
+    handler = Mangum(app, dsn=dsn)
     response = handler(mock_ws_send_event, {})
     assert response == {"statusCode": 500}
+
+
+@pytest.mark.parametrize(
+    "dsn", ["???://unknown/", "postgresql://", None, "http://localhost"]
+)
+def test_invalid_dsn(mock_ws_connect_event, mock_websocket_app, dsn):
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    with pytest.raises(ConfigurationError):
+        handler(mock_ws_connect_event, {})
+
+
+def test_sqlite_3_backend(
+    tmp_path,
+    mock_ws_connect_event,
+    mock_ws_send_event,
+    mock_ws_disconnect_event,
+    mock_websocket_app,
+) -> None:
+    dsn = f"sqlite://{tmp_path}/mangum.sqlite3"
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    response = handler(mock_ws_connect_event, {})
+    assert response == {"statusCode": 200}
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
+        send.return_value = None
+        response = handler(mock_ws_send_event, {})
+        assert response == {"statusCode": 200}
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    response = handler(mock_ws_disconnect_event, {})
+    assert response == {"statusCode": 200}
+
+
+@mock_dynamodb2
+def test_dynamodb_backend(
+    mock_ws_connect_event,
+    mock_ws_send_event,
+    mock_ws_disconnect_event,
+    mock_websocket_app,
+) -> None:
+    table_name = "mangum"
+    region_name = "ap-southeast-1"
+    dynamodb_resource = boto3.resource("dynamodb", region_name=region_name)
+    dynamodb_resource.meta.client.create_table(
+        TableName=table_name,
+        KeySchema=[{"AttributeName": "connectionId", "KeyType": "HASH"}],
+        AttributeDefinitions=[{"AttributeName": "connectionId", "AttributeType": "S"}],
+        ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
+    )
+
+    dsn = f"dynamodb://{table_name}?region={region_name}"
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    response = handler(mock_ws_connect_event, {})
+    assert response == {"statusCode": 200}
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
+        send.return_value = None
+        response = handler(mock_ws_send_event, {})
+        assert response == {"statusCode": 200}
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    response = handler(mock_ws_disconnect_event, {})
+    assert response == {"statusCode": 200}
+
+    table_does_not_exist = f"dynamodb://unknown-table?region={region_name}"
+    handler = Mangum(mock_websocket_app, dsn=table_does_not_exist)
+    with pytest.raises(WebSocketError):
+        response = handler(mock_ws_connect_event, {})
+
+    table = dynamodb_resource.Table(table_name)
+    table.delete_item(Key={"connectionId": "d4NsecoByQ0CH-Q="})
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    with pytest.raises(WebSocketError):
+        response = handler(mock_ws_send_event, {})
+
+    endpoint_dsn = f"dynamodb://{table_name}?region={region_name}&endpoint_url=http://localhost:5000"
+    handler = Mangum(mock_websocket_app, dsn=endpoint_dsn)
+    with pytest.raises(WebSocketError):
+        response = handler(mock_ws_connect_event, {})
+
+
+@pytest.mark.parametrize(
+    "dsn",
+    [
+        "s3://mangum-bucket-12345",
+        "s3://mangum-bucket-12345/",
+        "s3://mangum-bucket-12345/mykey/",
+        "s3://mangum-bucket-12345/mykey",
+    ],
+)
+@mock_s3
+def test_s3_backend(
+    tmp_path,
+    mock_ws_connect_event,
+    mock_ws_send_event,
+    mock_ws_disconnect_event,
+    mock_websocket_app,
+    dsn,
+) -> None:
+    # bucket = "mangum-bucket-12345"
+    # conn = boto3.resource("s3")
+    # conn.create_bucket(Bucket=bucket)
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    response = handler(mock_ws_connect_event, {})
+    assert response == {"statusCode": 200}
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
+        send.return_value = None
+        response = handler(mock_ws_send_event, {})
+        assert response == {"statusCode": 200}
+
+    handler = Mangum(mock_websocket_app, dsn=dsn)
+    response = handler(mock_ws_disconnect_event, {})
+    assert response == {"statusCode": 200}
+
+
+def test_postgresql_backend(
+    mock_ws_connect_event,
+    mock_ws_send_event,
+    mock_ws_disconnect_event,
+    mock_websocket_app,
+):
+    with testing.postgresql.Postgresql() as postgresql:
+        create_engine(postgresql.url())
+
+        dsn = postgresql.url()
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        response = handler(mock_ws_connect_event, {})
+        assert response == {"statusCode": 200}
+
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
+            send.return_value = None
+            response = handler(mock_ws_send_event, {})
+            assert response == {"statusCode": 200}
+
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        response = handler(mock_ws_disconnect_event, {})
+        assert response == {"statusCode": 200}
+
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        response = handler(mock_ws_connect_event, {})
+        assert response == {"statusCode": 200}
+
+
+def test_redis_backend(
+    mock_ws_connect_event,
+    mock_ws_send_event,
+    mock_ws_disconnect_event,
+    mock_websocket_app,
+):
+    with testing.redis.RedisServer() as redis_server:
+        _dsn = redis_server.dsn()
+        dsn = f"redis://{_dsn['host']}:{_dsn['port']}/{_dsn['db']}"
+
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        response = handler(mock_ws_connect_event, {})
+        assert response == {"statusCode": 200}
+
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        with mock.patch("mangum.websocket.WebSocket.post_to_connection") as send:
+            send.return_value = None
+            response = handler(mock_ws_send_event, {})
+            assert response == {"statusCode": 200}
+
+        handler = Mangum(mock_websocket_app, dsn=dsn)
+        response = handler(mock_ws_disconnect_event, {})
+        assert response == {"statusCode": 200}