django.db connection Example Code

connection is a callable within the django.db module of the Django project.

Example 1 from django-guardian

django-guardian (project documentation and PyPI page) provides per-object permissions in Django projects by enhancing the existing authentication backend. The project's code is open source under the MIT license.

django-guardian / guardian / shortcuts.py

# shortcuts.py
import warnings
from collections import defaultdict
from functools import partial
from itertools import groupby

from django.apps import apps
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group, Permission
from django.contrib.contenttypes.models import ContentType
from django.db import connection
from django.db.models import Count, Q, QuerySet
from django.shortcuts import _get_queryset
from django.db.models.expressions import Value
from django.db.models.functions import Cast, Replace
from django.db.models import (
    AutoField,
    BigIntegerField,
    CharField,
    ForeignKey,
    IntegerField,
    PositiveIntegerField,
    PositiveSmallIntegerField,
    SmallIntegerField,
    UUIDField,
)
from guardian.core import ObjectPermissionChecker
from guardian.ctypes import get_content_type
from guardian.exceptions import MixedContentTypeError, WrongAppError, MultipleIdentityAndObjectError
from guardian.utils import get_anonymous_user, get_group_obj_perms_model, get_identity, get_user_obj_perms_model
GroupObjectPermission = get_group_obj_perms_model()
UserObjectPermission = get_user_obj_perms_model()


def assign_perm(perm, user_or_group, obj=None):


## ... source file abbreviated to get to connection examples ...



    values = values.values_list(field_pk, flat=True)
    return queryset.filter(pk__in=values)


def _handle_pk_field(queryset):
    pk = queryset.model._meta.pk

    if isinstance(pk, ForeignKey):
        return _handle_pk_field(pk.target_field)

    if isinstance(
        pk,
        (
            IntegerField,
            AutoField,
            BigIntegerField,
            PositiveIntegerField,
            PositiveSmallIntegerField,
            SmallIntegerField,
        ),
    ):
        return partial(Cast, output_field=BigIntegerField())

    if isinstance(pk, UUIDField):
        if connection.features.has_native_uuid_field:
            return partial(Cast, output_field=UUIDField())
        return partial(
            Replace,
            text=Value('-'),
            replacement=Value(''),
            output_field=CharField(),
        )

    return None



## ... source file continues with no further connection examples...

Example 2 from django-push-notifications

django-push-notifications is a Django app for storing and interacting with push notification services such as Google's Firebase Cloud Messaging and Apple Notifications. The django-push-notification project's source code is available open source under the MIT license.

django-push-notifications / push_notifications / fields.py

# fields.py
import re
import struct

from django import forms
from django.core.validators import MaxValueValidator, MinValueValidator, RegexValidator
from django.db import connection, models
from django.utils.translation import ugettext_lazy as _


__all__ = ["HexadecimalField", "HexIntegerField"]

UNSIGNED_64BIT_INT_MIN_VALUE = 0
UNSIGNED_64BIT_INT_MAX_VALUE = 2 ** 64 - 1


hex_re = re.compile(r"^(([0-9A-f])|(0x[0-9A-f]))+$")
signed_integer_engines = [
    "django.db.backends.postgresql",
    "django.db.backends.postgresql_psycopg2",
    "django.contrib.gis.db.backends.postgis",
    "django.db.backends.sqlite3"
]


def _using_signed_storage():
  return connection.settings_dict["ENGINE"] in signed_integer_engines


def _signed_to_unsigned_integer(value):
    return struct.unpack("Q", struct.pack("q", value))[0]


def _unsigned_to_signed_integer(value):
    return struct.unpack("q", struct.pack("Q", value))[0]


def _hex_string_to_unsigned_integer(value):
    return int(value, 16)


def _unsigned_integer_to_hex_string(value):
    return hex(value).rstrip("L")


class HexadecimalField(forms.CharField):
    def __init__(self, *args, **kwargs):
        self.default_validators = [
            RegexValidator(hex_re, _("Enter a valid hexadecimal number"), "invalid")
        ]
        super(HexadecimalField, self).__init__(*args, **kwargs)

    def prepare_value(self, value):
        if value and not isinstance(value, str) \
          and connection.vendor in ("mysql", "sqlite"):
            value = _unsigned_integer_to_hex_string(value)
        return super(forms.CharField, self).prepare_value(value)


class HexIntegerField(models.BigIntegerField):

    validators = [
        MinValueValidator(UNSIGNED_64BIT_INT_MIN_VALUE),
        MaxValueValidator(UNSIGNED_64BIT_INT_MAX_VALUE)
    ]

    def db_type(self, connection):
      engine = connection.settings_dict["ENGINE"]
        if "mysql" in engine:
            return "bigint unsigned"
        elif "sqlite" in engine:
            return "UNSIGNED BIG INT"
        else:
            return super(HexIntegerField, self).db_type(connection=connection)

    def get_prep_value(self, value):
        if value is None or value == "":
            return None
        if isinstance(value, str):
            value = _hex_string_to_unsigned_integer(value)
        if _using_signed_storage():
            value = _unsigned_to_signed_integer(value)
        return value

    def from_db_value(self, value, *args):
        if value is None:
            return value
        if _using_signed_storage():
            value = _signed_to_unsigned_integer(value)
        return value

    def to_python(self, value):


## ... source file continues with no further connection examples...

Example 3 from django-rest-framework

Django REST Framework (project homepage and documentation, PyPI package information and more resources on Full Stack Python), often abbreviated as "DRF", is a popular Django extension for building web APIs. The project has fantastic documentation and a wonderful quickstart that serve as examples of how to make it easier for newcomers to get started.

The project is open sourced under the Encode OSS Ltd. license.

django-rest-framework / rest_framework / views.py

# views.py
from django.conf import settings
from django.core.exceptions import PermissionDenied
from django.db import connection, models, transaction
from django.http import Http404
from django.http.response import HttpResponseBase
from django.utils.cache import cc_delim_re, patch_vary_headers
from django.utils.encoding import smart_str
from django.views.decorators.csrf import csrf_exempt
from django.views.generic import View

from rest_framework import exceptions, status
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.schemas import DefaultSchema
from rest_framework.settings import api_settings
from rest_framework.utils import formatting


def get_view_name(view):
    name = getattr(view, 'name', None)
    if name is not None:
        return name

    name = view.__class__.__name__
    name = formatting.remove_trailing_string(name, 'View')
    name = formatting.remove_trailing_string(name, 'ViewSet')
    name = formatting.camelcase_to_spaces(name)

    suffix = getattr(view, 'suffix', None)
    if suffix:
        name += ' ' + suffix

    return name


def get_view_description(view, html=False):
    description = getattr(view, 'description', None)
    if description is None:
        description = view.__class__.__doc__ or ''

    description = formatting.dedent(smart_str(description))
    if html:
        return formatting.markup_description(description)
    return description


def set_rollback():
    atomic_requests = connection.settings_dict.get('ATOMIC_REQUESTS', False)
    if atomic_requests and connection.in_atomic_block:
        transaction.set_rollback(True)


def exception_handler(exc, context):
    if isinstance(exc, Http404):
        exc = exceptions.NotFound()
    elif isinstance(exc, PermissionDenied):
        exc = exceptions.PermissionDenied()

    if isinstance(exc, exceptions.APIException):
        headers = {}
        if getattr(exc, 'auth_header', None):
            headers['WWW-Authenticate'] = exc.auth_header
        if getattr(exc, 'wait', None):
            headers['Retry-After'] = '%d' % exc.wait

        if isinstance(exc.detail, (list, dict)):
            data = exc.detail
        else:
            data = {'detail': exc.detail}

        set_rollback()
        return Response(data, status=exc.status_code, headers=headers)



## ... source file continues with no further connection examples...

Example 4 from django-sql-explorer

django-sql-explorer (PyPI page), also referred to as "SQL Explorer", is a code library for the Django Admin that allows approved, authenticated users to view and execute direct database SQL queries. The tool keeps track of executed queries so users can share them with each other, as well as export results to downloadable formats. django-sql-explorer is provided as open source under the MIT license.

django-sql-explorer / explorer / apps.py

# apps.py
from django.apps import AppConfig
from django.db import connections as djcs
from django.core.exceptions import ImproperlyConfigured


class ExplorerAppConfig(AppConfig):

    name = 'explorer'

    def ready(self):
        from explorer.schema import build_async_schemas
        _validate_connections()
        build_async_schemas()


def _get_default():
    from explorer.app_settings import EXPLORER_DEFAULT_CONNECTION
    return EXPLORER_DEFAULT_CONNECTION


def _get_explorer_connections():
    from explorer.app_settings import EXPLORER_CONNECTIONS
    return EXPLORER_CONNECTIONS


def _validate_connections():

    if _get_default() not in _get_explorer_connections().values():
        raise ImproperlyConfigured(
            'EXPLORER_DEFAULT_CONNECTION is %s, but that alias is not present in the values of EXPLORER_CONNECTIONS'
            % _get_default())

    for name, conn_name in _get_explorer_connections().items():
        if conn_name not in djcs:
            raise ImproperlyConfigured(
                'EXPLORER_CONNECTIONS contains (%s, %s), but %s is not a valid Django DB connection.'
                % (name, conn_name, conn_name))



## ... source file continues with no further connection examples...

Example 5 from django-taggit

django-taggit (PyPI page) provides a way to create, store, manage and use tags in a Django project. The code for django-taggit is open source and maintained by the collaborative developer community group Jazzband.

django-taggit / taggit / managers.py

# managers.py
import uuid
from operator import attrgetter

from django import VERSION
from django.conf import settings
from django.contrib.contenttypes.fields import GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.db import connections, models, router
from django.db.models import signals
from django.db.models.fields.related import (
    ManyToManyRel,
    OneToOneRel,
    RelatedField,
    lazy_related_operation,
)
from django.db.models.query_utils import PathInfo
from django.utils.text import capfirst
from django.utils.translation import gettext_lazy as _

from taggit.forms import TagField
from taggit.models import (
    CommonGenericTaggedItemBase,
    GenericUUIDTaggedItemBase,
    TaggedItem,
)
from taggit.utils import require_instance_manager


class ExtraJoinRestriction:

    contains_aggregate = False



## ... source file abbreviated to get to connection examples ...


            kwargs = extra_filters if extra_filters else {}
            return self.through.tags_for(self.model, self.instance, **kwargs)

    def get_prefetch_queryset(self, instances, queryset=None):
        if queryset is not None:
            raise ValueError("Custom queryset can't be used for this lookup.")

        instance = instances[0]
        db = self._db or router.db_for_read(type(instance), instance=instance)

        fieldname = (
            "object_id"
            if issubclass(self.through, CommonGenericTaggedItemBase)
            else "content_object"
        )
        fk = self.through._meta.get_field(fieldname)
        query = {
            "%s__%s__in"
            % (self.through.tag_relname(), fk.name): {
                obj._get_pk_val() for obj in instances
            }
        }
        join_table = self.through._meta.db_table
        source_col = fk.column
        connection = connections[db]
        qn = connection.ops.quote_name
        qs = (
            self.get_queryset(query)
            .using(db)
            .extra(
                select={
                    "_prefetch_related_val": "{}.{}".format(
                        qn(join_table), qn(source_col)
                    )
                }
            )
        )

        if issubclass(self.through, GenericUUIDTaggedItemBase):

            def uuid_rel_obj_attr(v):
                value = attrgetter("_prefetch_related_val")(v)
                if value is not None and not isinstance(value, uuid.UUID):
                    input_form = "int" if isinstance(value, int) else "hex"
                    value = uuid.UUID(**{input_form: value})
                return value

            rel_obj_attr = uuid_rel_obj_attr
        else:
            rel_obj_attr = attrgetter("_prefetch_related_val")


## ... source file continues with no further connection examples...

Sponsored By

Sentry logo

Software errors are inevitable. Chaos is not. Try Sentry for free.


AssemblyAI logo

The most accurate speech-to-text API. Built for Python developers.

1. Introduction 2. Development Environments 3. Data 4. Web Development 5. Deployment 6. DevOps Changelog What Full Stack Means About the Author Future Directions Page Statuses Django ExtensionsDjango Example Codedjango.apps.config AppConfigdjango.conf settingsdjango.conf.urls.urldjango.contrib.admindjango.contrib.admin.filters SimpleListFilterdjango.contrib.admin.sites registerdjango.contrib.admin helpersdjango.contrib.admin.helpers ActionFormdjango.contrib.admin.helpers AdminFormdjango.contrib.admin.options IS_POPUP_VARdjango.contrib.admin.options IncorrectLookupParametersdjango.contrib.admin.options ModelAdmindjango.contrib.admin.options csrf_protect_mdjango.contrib.admin.sites NotRegistereddjango.contrib.admin.sites sitedjango.contrib.staticfiles findersdjango.contrib.staticfiles storagedjango.contrib.staticfiles.finders BaseFinderdjango.contrib.staticfiles.finders BaseStorageFinderdjango.contrib.staticfiles.finders finddjango.contrib.staticfiles.finders get_findersdjango.contrib.staticfiles.handlers StaticFilesHandlerdjango.contrib.staticfiles.storage CachedStaticFilesStoragedjango.contrib.staticfiles.storage HashedFilesMixindjango.contrib.staticfiles.storage ManifestStaticFilesStoragedjango.contrib.staticfiles.storage StaticFilesStoragedjango.contrib.staticfiles.storage staticfiles_storagedjango.contrib.staticfiles.utils matches_patternsdjango.core cachedjango.core checksdjango.core exceptionsdjango.core maildjango.core managementdjango.core serializersdjango.core signalsdjango.core signingdjango.core validatorsdjango.core.exceptions DisallowedRedirectdjango.core.exceptions FieldDoesNotExistdjango.core.exceptions FieldErrordjango.core.exceptions MiddlewareNotUseddjango.core.exceptions NON_FIELD_ERRORSdjango.core.exceptions ObjectDoesNotExistdjango.core.exceptions PermissionDenieddjango.core.exceptions SuspiciousFileOperationdjango.core.exceptions SuspiciousMultipartFormdjango.core.exceptions ValidationErrordjango.db DEFAULT_DB_ALIASdjango.db DataErrordjango.db DatabaseErrordjango.db IntegrityErrordjango.db ProgrammingErrordjango.db connectiondjango.db connectionsdjango.db migrationsdjango.db modelsdjango.db routerdjango.db transactiondjango.db.backends utilsdjango.db.migrations RunPythondjango.db.migrations.autodetector MigrationAutodetectordjango.db.migrations.exceptions IrreversibleErrordjango.db.migrations.executor MigrationExecutordjango.db.migrations.loader MIGRATIONS_MODULE_NAMEdjango.db.migrations.loader MigrationLoaderdjango.db.migrations.operations.base Operationdjango.db.migrations.state ProjectStatedjango.db.models.query BaseIterabledjango.db.models.query EmptyQuerySetdjango.db.models.query ModelIterabledjango.db.models.query Prefetchdjango.db.models.query Qdjango.db.models.query QuerySetdjango.db.models.query prefetch_related_objectsdjango.db.models.query_utils DeferredAttributedjango.db.models.query_utils PathInfodjango.db.models.query_utils Qdjango.db.models.signals post_deletedjango.db.models.signals post_savedjango.db.models.signals pre_deletedjango.db.models.signals pre_savedjango.forms BaseFormdjango.forms CheckboxInputdjango.forms CheckboxSelectMultipledjango.forms DateInputdjango.forms Fielddjango.forms FileInputdjango.forms FilePathFielddjango.forms Formdjango.forms HiddenInputdjango.forms ImageFielddjango.forms Mediadjango.forms MediaDefiningClassdjango.forms ModelChoiceFielddjango.forms ModelFormdjango.forms ModelMultipleChoiceFielddjango.forms MultipleChoiceFielddjango.forms Selectdjango.forms SelectMultipledjango.forms ValidationErrordjango.shortcuts get_list_or_404django.shortcuts get_object_or_404django.shortcuts redirectdjango.shortcuts renderdjango.shortcuts resolve_urldjango.template.base Contextdjango.template.base FilterExpressiondjango.template.base Nodedjango.template.base NodeListdjango.template.base Parserdjango.template.base Templatedjango.template.base TemplateSyntaxErrordjango.template.base TextNodedjango.template.base Tokendjango.template.base TokenTypedjango.template.base VariableDoesNotExistdjango.template.base VariableNodedjango.template.base token_kwargsdjango.urls URLPatterndjango.urls URLResolverdjango.urls clear_url_cachesdjango.urls get_callabledjango.urls get_resolverdjango.urls get_script_prefixdjango.urls includedjango.urls re_pathdjango.urls register_converterdjango.urls resolvedjango.urls reversedjango.utils dateformatdjango.utils dateparsedjango.utils datetime_safedjango.utils formatsdjango.utils module_loadingdjango.utils termcolorsdjango.utils translationdjango.utils treedjango.utils.cache add_never_cache_headersdjango.utils.cache cc_delim_redjango.utils.cache patch_cache_controldjango.utils.cache patch_response_headersdjango.utils.cache patch_vary_headersdjango.utils.crypto constant_time_comparedjango.utils.crypto get_random_stringdjango.utils.datastructures MultiValueDictdjango.utils.dateparse parse_datetimedjango.utils.dateparse parse_durationdjango.utils.dates MONTHSdjango.utils.datetime_safe datetimedjango.utils.decorators method_decoratordjango.utils.deprecation MiddlewareMixindjango.utils.deprecation RenameMethodsBasedjango.utils.duration duration_stringdjango.utils.encoding DjangoUnicodeDecodeErrordjango.utils.encoding filepath_to_uridjango.utils.encoding force_bytesdjango.utils.encoding force_strdjango.utils.encoding force_textdjango.utils.encoding iri_to_uridjango.utils.encoding is_protected_typedjango.utils.encoding smart_bytesdjango.utils.encoding smart_strdjango.utils.encoding smart_textdjango.utils.encoding uri_to_iridjango.utils.formats get_formatdjango.utils.formats localize_inputdjango.utils.formats sanitize_separatorsdjango.utils.functional LazyObjectdjango.utils.functional Promisedjango.utils.functional SimpleLazyObjectdjango.utils.functional keep_lazydjango.utils.functional lazydjango.utils.functional total_orderingdjango.utils.functional wrapsdjango.utils.html conditional_escapedjango.utils.html escapedjango.utils.html escapejsdjango.utils.html format_html_joindjango.utils.html mark_safedjango.utils.html smart_urlquotedjango.utils.html strip_tagsdjango.utils.http base36_to_intdjango.utils.http http_datedjango.utils.http int_to_base36django.utils.http is_safe_urldjango.utils.http unquotedjango.utils.http url_has_allowed_host_and_schemedjango.utils.http urlencodedjango.utils.http urlquotedjango.utils.http urlunquotedjango.utils.ipv6 clean_ipv6_addressdjango.utils.itercompat is_iterabledjango.utils.module_loading autodiscover_modulesdjango.utils.module_loading import_stringdjango.utils.module_loading module_has_submoduledjango.utils.numberformat formatdjango.utils.safestring SafeDatadjango.utils.safestring SafeTextdjango.utils.safestring mark_safedjango.utils.termcolors colorizedjango.utils.text Truncatordjango.utils.text capfirstdjango.utils.text format_lazydjango.utils.text get_text_listdjango.utils.text get_valid_filenamedjango.utils.text slugifydjango.utils.timezone get_current_timezonedjango.utils.timezone make_awaredjango.utils.timezone nowdjango.utils.timezone timedeltadjango.utils.translation LANGUAGE_SESSION_KEYdjango.utils.translation activatedjango.utils.translation deactivate_alldjango.utils.translation get_languagedjango.utils.translation get_language_from_requestdjango.utils.translation gettextdjango.utils.translation gettext_lazydjango.utils.translation ngettextdjango.utils.translation overridedjango.utils.translation pgettextdjango.utils.translation pgettext_lazydjango.utils.translation ugettextdjango.utils.translation ugettext_lazydjango.utils.translation ungettextdjango.utils.translation ungettext_lazydjango.utils.version get_complete_versiondjango.views csrfdjango.views.debug get_default_exception_reporter_filterdjango.views.decorators.csrf csrf_exemptdjango.views.decorators.debug sensitive_post_parametersdjango.views.decorators.http require_GETdjango.views.decorators.http require_POSTdjango.views.generic CreateViewdjango.views.generic DeleteViewdjango.views.generic DetailViewdjango.views.generic FormViewdjango.views.generic ListViewdjango.views.generic RedirectViewdjango.views.generic TemplateViewdjango.views.generic UpdateViewdjango.views.generic Viewdjango.views.generic.base RedirectViewdjango.views.generic.base TemplateResponseMixindjango.views.generic.base TemplateViewdjango.views.generic.base Viewdjango.views.generic.detail SingleObjectMixindjango.views.generic.edit CreateViewdjango.views.generic.edit DeleteViewdjango.views.generic.edit DeletionMixindjango.views.generic.edit FormMixindjango.views.generic.edit FormViewdjango.views.generic.list ListViewdjango.views.generic.list MultipleObjectMixindjango.views.i18n JavaScriptCatalogdjango.views.static servedjango.views.static was_modified_sincedjango.contrib.auth.decorators login_requireddjango.contrib.auth get_user_modeldjango.contrib.auth.hashers make_passworddjango.core.exceptions ImproperlyConfigureddjango.core.mail.messages EmailMessagedjango.core.mail.send_maildjango.core.management.base BaseCommanddjango.db.models AutoFielddjango.db.models BooleanFielddjango.db.models CharFielddjango.db.models DateFielddjango.db.models DateTimeFielddjango.db.models FileFielddjango.db.models ForeignKeydjango.db.models GenericIPAddressFielddjango.db.models ImageFielddjango.db.models IntegerFielddjango.db.models Modeldjango.db.models PositiveIntegerFielddjango.db.models PositiveSmallIntegerFielddjango.db.models.signaldjango.db.models SlugFielddjango.db.models SmallIntegerFielddjango.db.models TextFielddjango.db OperationalErrordjango.dispatch Signaldjango.formsdjango.forms BooleanFielddjango.forms CharFielddjango.forms ChoiceFielddjango.forms DateFielddjango.forms DateTimeFielddjango.forms EmailFielddjango.forms IntegerFielddjango.forms TypedChoiceFielddjango.http Http404django.http HttpResponsedjango.http HttpResponseBadRequestdjango.http HttpResponseForbiddendjango.http HttpResponseNotModifieddjango.http HttpResponsePermanentRedirectdjango.http HttpResponseRedirectdjango.template.response SimpleTemplateResponsedjango.template.response TemplateResponsedjango.urls.pathdjango.urls reverse_lazydjango.urls.exceptions NoReverseMatchdjango.urls.exceptions Resolver404django.utils.html format_htmldjango.utils.timezone ...or view the full table of contents.

Full Stack Python

Full Stack Python is an open book that explains concepts in plain language and provides helpful resources for those topics.
Updates via Twitter & Facebook.

Matt Makai 2012-2020