diff --git a/django/db/__init__.py b/django/db/__init__.py
index 8395468..30e1c3d 100644
--- a/django/db/__init__.py
+++ b/django/db/__init__.py
@@ -22,9 +22,19 @@ router = ConnectionRouter(settings.DATABASE_ROUTERS)
 # we manually create the dictionary from the settings, passing only the
 # settings that the database backends care about. Note that TIME_ZONE is used
 # by the PostgreSQL backends.
-# we load all these up for backwards compatibility, you should use
+# We load all these up for backwards compatibility, you should use
 # connections['default'] instead.
-connection = connections[DEFAULT_DB_ALIAS]
+class DefaultConnectionProxy(object):
+    """
+    Proxy for the thread-local default connection.
+    """
+    def __getattr__(self, item):
+        return getattr(connections[DEFAULT_DB_ALIAS], item)
+
+    def __setattr__(self, name, value):
+        return setattr(connections[DEFAULT_DB_ALIAS], name, value)
+
+connection = DefaultConnectionProxy()
 backend = load_backend(connection.settings_dict['ENGINE'])
 
 # Register an event that closes the database connection
diff --git a/django/db/backends/__init__.py b/django/db/backends/__init__.py
index f2bde84..460746b 100644
--- a/django/db/backends/__init__.py
+++ b/django/db/backends/__init__.py
@@ -1,8 +1,9 @@
+from django.db.utils import DatabaseError
+
 try:
     import thread
 except ImportError:
     import dummy_thread as thread
-from threading import local
 from contextlib import contextmanager
 
 from django.conf import settings
@@ -13,14 +14,15 @@ from django.utils.importlib import import_module
 from django.utils.timezone import is_aware
 
 
-class BaseDatabaseWrapper(local):
+class BaseDatabaseWrapper(object):
     """
     Represents a database connection.
     """
     ops = None
     vendor = 'unknown'
 
-    def __init__(self, settings_dict, alias=DEFAULT_DB_ALIAS):
+    def __init__(self, settings_dict, alias=DEFAULT_DB_ALIAS,
+                 check_same_thread=True):
         # `settings_dict` should be a dictionary containing keys such as
         # NAME, USER, etc. It's called `settings_dict` instead of `settings`
         # to disambiguate it from Django settings modules.
@@ -34,6 +36,8 @@ class BaseDatabaseWrapper(local):
         self.transaction_state = []
         self.savepoint_state = 0
         self._dirty = None
+        self.thread_id = thread.get_ident()
+        self.check_same_thread = check_same_thread
 
     def __eq__(self, other):
         return self.alias == other.alias
@@ -116,6 +120,17 @@ class BaseDatabaseWrapper(local):
                 "pending COMMIT/ROLLBACK")
         self._dirty = False
 
+    def validate_thread(self):
+        if self.check_same_thread:
+            current_thread_id = thread.get_ident()
+            if current_thread_id != self.thread_id:
+                raise DatabaseError ("DatabaseWrapper objects created in a "
+                    "thread can only be used in that same thread, unless you "
+                    "explicitly set its 'check_same_thread' property to "
+                    "False. The object was created in thread id %s and this "
+                    "is thread id %s" % (self.thread_id, current_thread_id))
+        return True
+
     def is_dirty(self):
         """
         Returns True if the current transaction requires a commit for changes to
@@ -179,6 +194,7 @@ class BaseDatabaseWrapper(local):
         """
         Commits changes if the system is not in managed transaction mode.
         """
+        self.validate_thread()
         if not self.is_managed():
             self._commit()
             self.clean_savepoints()
@@ -189,6 +205,7 @@ class BaseDatabaseWrapper(local):
         """
         Rolls back changes if the system is not in managed transaction mode.
         """
+        self.validate_thread()
         if not self.is_managed():
             self._rollback()
         else:
@@ -198,6 +215,7 @@ class BaseDatabaseWrapper(local):
         """
         Does the commit itself and resets the dirty flag.
         """
+        self.validate_thread()
         self._commit()
         self.set_clean()
 
@@ -205,6 +223,7 @@ class BaseDatabaseWrapper(local):
         """
         This function does the rollback itself and resets the dirty flag.
         """
+        self.validate_thread()
         self._rollback()
         self.set_clean()
 
@@ -228,6 +247,7 @@ class BaseDatabaseWrapper(local):
         Rolls back the most recent savepoint (if one exists). Does nothing if
         savepoints are not supported.
         """
+        self.validate_thread()
         if self.savepoint_state:
             self._savepoint_rollback(sid)
 
@@ -236,6 +256,7 @@ class BaseDatabaseWrapper(local):
         Commits the most recent savepoint (if one exists). Does nothing if
         savepoints are not supported.
         """
+        self.validate_thread()
         if self.savepoint_state:
             self._savepoint_commit(sid)
 
@@ -269,11 +290,13 @@ class BaseDatabaseWrapper(local):
         pass
 
     def close(self):
+        self.validate_thread()
         if self.connection is not None:
             self.connection.close()
             self.connection = None
 
     def cursor(self):
+        self.validate_thread()
         if (self.use_debug_cursor or
             (self.use_debug_cursor is None and settings.DEBUG)):
             cursor = self.make_debug_cursor(self._cursor())
diff --git a/django/db/backends/sqlite3/base.py b/django/db/backends/sqlite3/base.py
index a610606..4a96c7c 100644
--- a/django/db/backends/sqlite3/base.py
+++ b/django/db/backends/sqlite3/base.py
@@ -220,8 +220,8 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         'iendswith': "LIKE %s ESCAPE '\\'",
     }
 
-    def __init__(self, *args, **kwargs):
-        super(DatabaseWrapper, self).__init__(*args, **kwargs)
+    def __init__(self, settings_dict, *args, **kwargs):
+        super(DatabaseWrapper, self).__init__(settings_dict, *args, **kwargs)
 
         self.features = DatabaseFeatures(self)
         self.ops = DatabaseOperations(self)
@@ -229,6 +229,8 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         self.creation = DatabaseCreation(self)
         self.introspection = DatabaseIntrospection(self)
         self.validation = BaseDatabaseValidation(self)
+        self.check_same_thread = (
+            settings_dict['OPTIONS'].get('check_same_thread'), True)
 
     def _cursor(self):
         if self.connection is None:
@@ -239,6 +241,7 @@ class DatabaseWrapper(BaseDatabaseWrapper):
             kwargs = {
                 'database': settings_dict['NAME'],
                 'detect_types': Database.PARSE_DECLTYPES | Database.PARSE_COLNAMES,
+                'check_same_thread': 1 if self.check_same_thread else 0,
             }
             kwargs.update(settings_dict['OPTIONS'])
             self.connection = Database.connect(**kwargs)
diff --git a/django/db/utils.py b/django/db/utils.py
index f0c13e3..41ad6df 100644
--- a/django/db/utils.py
+++ b/django/db/utils.py
@@ -1,4 +1,5 @@
 import os
+from threading import local
 
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
@@ -50,7 +51,7 @@ class ConnectionDoesNotExist(Exception):
 class ConnectionHandler(object):
     def __init__(self, databases):
         self.databases = databases
-        self._connections = {}
+        self._connections = local()
 
     def ensure_defaults(self, alias):
         """
@@ -73,16 +74,19 @@ class ConnectionHandler(object):
             conn.setdefault(setting, None)
 
     def __getitem__(self, alias):
-        if alias in self._connections:
-            return self._connections[alias]
+        if hasattr(self._connections, alias):
+            return getattr(self._connections, alias)
 
         self.ensure_defaults(alias)
         db = self.databases[alias]
         backend = load_backend(db['ENGINE'])
         conn = backend.DatabaseWrapper(db, alias)
-        self._connections[alias] = conn
+        setattr(self._connections, alias, conn)
         return conn
 
+    def __setitem__(self, key, value):
+        setattr(self._connections, key, value)
+
     def __iter__(self):
         return iter(self.databases)
 
diff --git a/tests/regressiontests/backends/tests.py b/tests/regressiontests/backends/tests.py
index 936f010..998e7ce 100644
--- a/tests/regressiontests/backends/tests.py
+++ b/tests/regressiontests/backends/tests.py
@@ -3,6 +3,7 @@
 from __future__ import with_statement, absolute_import
 
 import datetime
+import threading
 
 from django.conf import settings
 from django.core.management.color import no_style
@@ -283,7 +284,7 @@ class ConnectionCreatedSignalTest(TestCase):
         connection_created.connect(receiver)
         connection.close()
         cursor = connection.cursor()
-        self.assertTrue(data["connection"] is connection)
+        self.assertTrue(data["connection"].connection is connection.connection)
 
         connection_created.disconnect(receiver)
         data.clear()
@@ -446,3 +447,90 @@ class FkConstraintsTests(TransactionTestCase):
                         connection.check_constraints()
             finally:
                 transaction.rollback()
+
+class ThreadTests(TestCase):
+
+    def test_default_connection_thread_local(self):
+        """
+        Ensure that the default connection (i.e. django.db.connection) is
+        different for each thread.
+        Refs #17258.
+        """
+        connections_set = set()
+        connection.cursor()
+        connections_set.add(connection.connection)
+        def runner():
+            from django.db import connection
+            connection.cursor()
+            connections_set.add(connection.connection)
+        for x in xrange(2):
+            t = threading.Thread(target=runner)
+            t.start()
+            t.join()
+        self.assertEquals(len(connections_set), 3)
+        # Finish by closing the connections opened by the other threads (the
+        # connection opened in the main thread will automatically be closed on
+        # teardown).
+        for conn in connections_set:
+            if conn != connection.connection:
+                conn.close()
+
+    def test_connections_thread_local(self):
+        """
+        Ensure that the connections are different for each thread.
+        Refs #17258.
+        """
+        connections_set = set()
+        for conn in connections.all():
+            connections_set.add(conn)
+        def runner():
+            from django.db import connections
+            for conn in connections.all():
+                connections_set.add(conn)
+        for x in xrange(2):
+            t = threading.Thread(target=runner)
+            t.start()
+            t.join()
+        self.assertEquals(len(connections_set), 6)
+        # Finish by closing the connections opened by the other threads (the
+        # connection opened in the main thread will automatically be closed on
+        # teardown).
+        for conn in connections_set:
+            if conn != connection:
+                conn.close()
+
+    def test_pass_connection_between_threads(self):
+        """
+        Ensure that one can pass a connection from one thread to the other.
+        Refs #17258.
+        """
+        models.Person.objects.create(first_name="John", last_name="Doe")
+
+        def do_thread():
+            def runner(main_thread_connection):
+                from django.db import connections
+                connections['default'] = main_thread_connection
+                try:
+                    models.Person.objects.get(first_name="John", last_name="Doe")
+                except DatabaseError, e:
+                    exceptions.append(e)
+            t = threading.Thread(target=runner, args=[connections['default']])
+            t.start()
+            t.join()
+
+        # By default (without touching check_same_thread)
+        exceptions = []
+        do_thread()
+        self.assertTrue(isinstance(exceptions[0], DatabaseError))
+
+        # If explicitly setting check_same_thread to True
+        connections['default'].check_same_thread = True
+        exceptions = []
+        do_thread()
+        self.assertTrue(isinstance(exceptions[0], DatabaseError))
+
+        # If explicitly setting check_same_thread to False
+        connections['default'].check_same_thread = False
+        exceptions = []
+        do_thread()
+        self.assertEqual(len(exceptions), 0)
\ No newline at end of file
diff --git a/tests/test_sqlite.py b/tests/test_sqlite.py
index de8bf93..c143edd 100644
--- a/tests/test_sqlite.py
+++ b/tests/test_sqlite.py
@@ -14,9 +14,15 @@
 
 DATABASES = {
     'default': {
-        'ENGINE': 'django.db.backends.sqlite3'
+        'ENGINE': 'django.db.backends.sqlite3',
+        'OPTIONS': {
+            'check_same_thread': False,
+        }
     },
     'other': {
         'ENGINE': 'django.db.backends.sqlite3',
+        'OPTIONS': {
+            'check_same_thread': False,
+        }
     }
 }
