Support for unix socket for aiomysql and asyncmy

databases/backends/asyncmy.py

@@ -40,6 +40,7 @@ def _get_connection_kwargs(self) -> dict:
         max_size = url_options.get("max_size")
         pool_recycle = url_options.get("pool_recycle")
         ssl = url_options.get("ssl")
+        unix_socket = url_options.get("unix_socket")
 
         if min_size is not None:
             kwargs["minsize"] = int(min_size)
@@ -49,6 +50,8 @@ def _get_connection_kwargs(self) -> dict:
             kwargs["pool_recycle"] = int(pool_recycle)
         if ssl is not None:
             kwargs["ssl"] = {"true": True, "false": False}[ssl.lower()]
+        if unix_socket is not None:
+            kwargs["unix_socket"] = unix_socket
 
         for key, value in self._options.items():
             # Coerce 'min_size' and 'max_size' for consistency.

databases/backends/mysql.py

@@ -40,6 +40,7 @@ def _get_connection_kwargs(self) -> dict:
         max_size = url_options.get("max_size")
         pool_recycle = url_options.get("pool_recycle")
         ssl = url_options.get("ssl")
+        unix_socket = url_options.get("unix_socket")
 
         if min_size is not None:
             kwargs["minsize"] = int(min_size)
@@ -49,6 +50,8 @@ def _get_connection_kwargs(self) -> dict:
             kwargs["pool_recycle"] = int(pool_recycle)
         if ssl is not None:
             kwargs["ssl"] = {"true": True, "false": False}[ssl.lower()]
+        if unix_socket is not None:
+            kwargs["unix_socket"] = unix_socket
 
         for key, value in self._options.items():
             # Coerce 'min_size' and 'max_size' for consistency.

tests/test_connection_options.py

@@ -77,6 +77,15 @@ def test_mysql_pool_size():
     assert kwargs == {"minsize": 1, "maxsize": 20}
 
 
+@pytest.mark.skipif(sys.version_info >= (3, 10), reason="requires python3.9 or lower")
+def test_mysql_unix_socket():
+    backend = MySQLBackend(
+        "mysql+aiomysql://username:password@/testsuite?unix_socket=/tmp/mysqld/mysqld.sock"
+    )
+    kwargs = backend._get_connection_kwargs()
+    assert kwargs == {"unix_socket": "/tmp/mysqld/mysqld.sock"}
+
+
 @pytest.mark.skipif(sys.version_info >= (3, 10), reason="requires python3.9 or lower")
 def test_mysql_explicit_pool_size():
     backend = MySQLBackend("mysql://localhost/database", min_size=1, max_size=20)
@@ -114,6 +123,15 @@ def test_asyncmy_pool_size():
     assert kwargs == {"minsize": 1, "maxsize": 20}
 
 
+@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7 or higher")
+def test_asyncmy_unix_socket():
+    backend = AsyncMyBackend(
+        "mysql+asyncmy://username:password@/testsuite?unix_socket=/tmp/mysqld/mysqld.sock"
+    )
+    kwargs = backend._get_connection_kwargs()
+    assert kwargs == {"unix_socket": "/tmp/mysqld/mysqld.sock"}
+
+
 @pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7 or higher")
 def test_asyncmy_explicit_pool_size():
     backend = AsyncMyBackend("mysql://localhost/database", min_size=1, max_size=20)

tests/test_database_url.py

@@ -69,6 +69,11 @@ def test_database_url_options():
     u = DatabaseURL("postgresql://localhost/mydatabase?pool_size=20&ssl=true")
     assert u.options == {"pool_size": "20", "ssl": "true"}
 
+    u = DatabaseURL(
+        "mysql+asyncmy://username:password@/testsuite?unix_socket=/tmp/mysqld/mysqld.sock"
+    )
+    assert u.options == {"unix_socket": "/tmp/mysqld/mysqld.sock"}
+
 
 def test_replace_database_url_components():
     u = DatabaseURL("postgresql://localhost/mydatabase")