adapters load a postgresql timestamptz using the a specific timezone the timezone can be none too in which case it will be chopped a range dumper customized for django def upgradeself obj format dump ranges containing naive datetimes as tstzrange because django doesn t use tzaware ones dumper super upgradeobj format if dumper is not self and dumper oid tsrangeoid dumper oid tstzrangeoid return dumper lrucache def getadapterstemplateusetz timezone create at adapters map extending the base one ctx adapt adaptersmapadapters register a noop dumper to avoid a round trip from psycopg version 3 decode to json dumps to json loads when using a custom decoder in jsonfield ctx registerloaderjsonb textloader don t convert automatically from postgresql network types to python ipaddress ctx registerloaderinet textloader ctx registerloadercidr textloader ctx registerdumperrange djangorangedumper register a timestamptz loader configured on self timezone this however can be overridden by createcursor registertzloadertimezone ctx return ctx ispsycopg3 true except importerror from enum import intenum from psycopg2 import errors extensions sql noqa from psycopg2 extras import daterange datetimerange datetimetzrange inet noqa from psycopg2 extras import json as jsonb noqa from psycopg2 extras import numericrange range noqa rangetypes daterange datetimerange datetimetzrange numericrange class isolationlevelintenum readuncommitted extensions isolationlevelreaduncommitted readcommitted extensions isolationlevelreadcommitted repeatableread extensions isolationlevelrepeatableread serializable extensions isolationlevelserializable def quotevalue connectionnone adapted extensions adaptvalue if hasattradapted encoding adapted encoding utf8 getquoted returns a quoted bytestring of the adapted value return adapted getquoted decode sql quote quote def mogrifysql params connection with connection cursor as cursor return cursor mogrifysql params decode ispsycopg3 false adapters load a postgresql timestamptz using the a specific timezone the timezone can be none too in which case it will be chopped a range dumper customized for django dump ranges containing naive datetimes as tstzrange because django doesn t use tz aware ones create at adapters map extending the base one register a no op dumper to avoid a round trip from psycopg version 3 decode to json dumps to json loads when using a custom decoder in jsonfield don t convert automatically from postgresql network types to python ipaddress register a timestamptz loader configured on self timezone this however can be overridden by create_cursor noqa noqa noqa noqa getquoted returns a quoted bytestring of the adapted value
import ipaddress
from functools import lru_cache

try:
    from psycopg import ClientCursor, IsolationLevel, adapt, adapters, errors, sql
    from psycopg.postgres import types
    from psycopg.types.datetime import TimestamptzLoader
    from psycopg.types.json import Jsonb
    from psycopg.types.range import Range, RangeDumper
    from psycopg.types.string import TextLoader

    Inet = ipaddress.ip_address

    DateRange = DateTimeRange = DateTimeTZRange = NumericRange = Range
    RANGE_TYPES = (Range,)

    TSRANGE_OID = types["tsrange"].oid
    TSTZRANGE_OID = types["tstzrange"].oid

    def mogrify(sql, params, connection):
        with connection.cursor() as cursor:
            return ClientCursor(cursor.connection).mogrify(sql, params)

    
    class BaseTzLoader(TimestamptzLoader):
        

        timezone = None

        def load(self, data):
            res = super().load(data)
            return res.replace(tzinfo=self.timezone)

    def register_tzloader(tz, context):
        class SpecificTzLoader(BaseTzLoader):
            timezone = tz

        context.adapters.register_loader("timestamptz", SpecificTzLoader)

    class DjangoRangeDumper(RangeDumper):
        

        def upgrade(self, obj, format):
            
            
            dumper = super().upgrade(obj, format)
            if dumper is not self and dumper.oid == TSRANGE_OID:
                dumper.oid = TSTZRANGE_OID
            return dumper

    @lru_cache
    def get_adapters_template(use_tz, timezone):
        
        ctx = adapt.AdaptersMap(adapters)
        
        
        
        ctx.register_loader("jsonb", TextLoader)
        
        
        ctx.register_loader("inet", TextLoader)
        ctx.register_loader("cidr", TextLoader)
        ctx.register_dumper(Range, DjangoRangeDumper)
        
        
        register_tzloader(timezone, ctx)
        return ctx

    is_psycopg3 = True

except ImportError:
    from enum import IntEnum

    from psycopg2 import errors, extensions, sql  
    from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange, Inet  
    from psycopg2.extras import Json as Jsonb  
    from psycopg2.extras import NumericRange, Range  

    RANGE_TYPES = (DateRange, DateTimeRange, DateTimeTZRange, NumericRange)

    class IsolationLevel(IntEnum):
        READ_UNCOMMITTED = extensions.ISOLATION_LEVEL_READ_UNCOMMITTED
        READ_COMMITTED = extensions.ISOLATION_LEVEL_READ_COMMITTED
        REPEATABLE_READ = extensions.ISOLATION_LEVEL_REPEATABLE_READ
        SERIALIZABLE = extensions.ISOLATION_LEVEL_SERIALIZABLE

    def _quote(value, connection=None):
        adapted = extensions.adapt(value)
        if hasattr(adapted, "encoding"):
            adapted.encoding = "utf8"
        
        return adapted.getquoted().decode()

    sql.quote = _quote

    def mogrify(sql, params, connection):
        with connection.cursor() as cursor:
            return cursor.mogrify(sql, params).decode()

    is_psycopg3 = False