diff --git a/django/db/backends/sqlite3/base.py b/django/db/backends/sqlite3/base.py index 1548dbb30743..120c90077c76 100644 --- a/django/db/backends/sqlite3/base.py +++ b/django/db/backends/sqlite3/base.py @@ -7,10 +7,11 @@ import warnings from collections.abc import Mapping from itertools import chain, tee +from pathlib import Path from sqlite3 import dbapi2 as Database from django.core.exceptions import ImproperlyConfigured -from django.db import IntegrityError +from django.db import IntegrityError, InterfaceError from django.db.backends.base.base import BaseDatabaseWrapper from django.utils.asyncio import async_unsafe from django.utils.dateparse import parse_date, parse_datetime, parse_time @@ -202,6 +203,11 @@ def get_database_version(self): @async_unsafe def get_new_connection(self, conn_params): + if not self.is_in_memory_db(): + try: + Path(conn_params["database"]).parent.mkdir(exist_ok=True, parents=True) + except OSError as e: + raise InterfaceError(f"Error creating path to SQLite database: {e}.") conn = Database.connect(**conn_params) register_functions(conn) diff --git a/tests/backends/sqlite/tests.py b/tests/backends/sqlite/tests.py index f47e96be4e34..9f7e7b7c9b93 100644 --- a/tests/backends/sqlite/tests.py +++ b/tests/backends/sqlite/tests.py @@ -11,6 +11,7 @@ from django.core.exceptions import ImproperlyConfigured from django.db import ( DEFAULT_DB_ALIAS, + InterfaceError, NotSupportedError, connection, connections, @@ -144,6 +145,74 @@ def test_init_command(self): finally: connections["default"]._close() + def test_nested_path_name(self): + with tempfile.TemporaryDirectory() as tmp: + settings_dict = { + "default": { + "ENGINE": "django.db.backends.sqlite3", + "NAME": Path(tmp) / "subdir" / "subsubdir" / "test.db", + }, + } + connections = ConnectionHandler(settings_dict) + connections["default"].ensure_connection() + connections["default"].close() + self.assertTrue( + os.path.isfile(os.path.join(tmp, "subdir", "subsubdir", "test.db")) + ) + + def test_nested_str_path_name(self): + with tempfile.TemporaryDirectory() as tmp: + settings_dict = { + "default": { + "ENGINE": "django.db.backends.sqlite3", + "NAME": os.path.join(tmp, "subdir", "test.db"), + }, + } + connections = ConnectionHandler(settings_dict) + connections["default"].ensure_connection() + connections["default"].close() + self.assertTrue(os.path.isfile(os.path.join(tmp, "subdir", "test.db"))) + + def test_database_name_with_file_in_path_not_allowed(self): + tests = ("subdir/test.db", "subdir/subsubdir/test.db") + for db_path in tests: + with self.subTest(db_path): + with tempfile.TemporaryDirectory() as tmp: + settings_dict = { + "default": { + "ENGINE": "django.db.backends.sqlite3", + "NAME": Path(tmp) / db_path, + }, + } + # write a file to the parent path + (Path(tmp) / "subdir").write_text("test") + connections = ConnectionHandler(settings_dict) + with self.assertRaisesMessage( + InterfaceError, + "Error creating path to SQLite database", + ): + connections["default"].ensure_connection() + + @mock.patch( + "django.db.backends.sqlite3.base.Path.mkdir", + side_effect=OSError(1, "A mock error"), + ) + def test_make_parent_path_oserror(self, mock_oserror): + with tempfile.TemporaryDirectory() as tmp: + settings_dict = { + "default": { + "ENGINE": "django.db.backends.sqlite3", + "NAME": Path(tmp) / "db" / "test.db", + }, + } + + connections = ConnectionHandler(settings_dict) + with self.assertRaisesMessage( + InterfaceError, + "Error creating path to SQLite database: [Errno 1] A mock error", + ): + connections["default"].ensure_connection() + @unittest.skipUnless(connection.vendor == "sqlite", "SQLite tests") @isolate_apps("backends")