pypy-postgresql / pypy / module / postgresql / interp_state.py

from pypy.interpreter.buffer import Buffer
from pypy.interpreter.error import OperationError
from pypy.interpreter.gateway import unwrap_spec
from pypy.module.postgresql import interp_types
from pypy.module.postgresql.interp_adapters import (W_QuotedString, W_AsIs,
    W_Float, W_Decimal, W_Boolean, W_Binary, W_List, W_DateTime)


def register_type(space, w_type, w_obj=None):
    from pypy.module.postgresql.interp_connection import W_Connection
    from pypy.module.postgresql.interp_cursor import W_Cursor

    if isinstance(w_obj, W_Connection):
        typecasts = w_obj.typecasts
    elif isinstance(w_obj, W_Cursor):
        typecasts = w_obj.typecasts
    elif space.is_w(w_obj, space.w_None):
        typecasts = get(space).typecasts
    else:
        raise OperationError(space.w_TypeError, space.wrap(
            "argument 2 must be a connection, cursor or None"
        ))
    _register_type(space, w_type, typecasts)

def _register_type(space, w_type, typecasts):
    w_type = space.interp_w(interp_types.W_Type, w_type)
    for value in w_type.values:
        typecasts[value] = w_type

@unwrap_spec(name=str)
def new_type(space, w_oids, name, w_adapt_func):
    values = []
    for w_item in space.listview(w_oids):
        values.append(space.int_w(w_item))
    return interp_types.W_Type(
        name,
        values,
        w_py_caster=w_adapt_func,
    )


class State(object):
    def __init__(self, space):
        self.w_Warning = space.new_exception_class("Warning", space.w_StandardError)
        self.w_Error = space.new_exception_class("Error", space.w_StandardError)
        self.w_InterfaceError = space.new_exception_class("InterfaceError", self.w_Error)
        self.w_DatabaseError = space.new_exception_class("DatabaseError", self.w_Error)
        self.w_DataError = space.new_exception_class("DataError", self.w_DatabaseError)
        self.w_OperationalError = space.new_exception_class("OperationalError", self.w_DatabaseError)
        self.w_IntegrityError = space.new_exception_class("IntegrityError", self.w_DatabaseError)
        self.w_InternalError = space.new_exception_class("InternalError", self.w_DatabaseError)
        self.w_ProgrammingError = space.new_exception_class("ProgrammingError", self.w_DatabaseError)
        self.w_NotSupportedError = space.new_exception_class("NotSupportedError", self.w_DatabaseError)

        self.adapters_w = {
            space.w_str: space.gettypefor(W_QuotedString),
            space.w_unicode: space.gettypefor(W_QuotedString),
            space.w_int: space.gettypefor(W_AsIs),
            space.w_long: space.gettypefor(W_AsIs),
            space.w_float: space.gettypefor(W_Float),
            space.w_bool: space.gettypefor(W_Boolean),
            space.gettypefor(Buffer): space.gettypefor(W_Binary),
            space.w_list: space.gettypefor(W_List),
        }

        # Values taken from psycopg2.
        w = space.wrap
        self.w_STRING = interp_types.W_Type(
            "STRING",
            [19, 18, 25, 1042, 1043],
            interp_types.cast_string
        )
        self.w_NUMBER = interp_types.W_Type(
            "NUMBER",
            [20, 33, 21, 701, 700, 1700],
            interp_types.cast_float,
        )
        self.w_LONGINTEGER = interp_types.W_Type(
            "LONGINTEGER",
            [20],
            interp_types.cast_longinteger
        )
        self.w_INTEGER = interp_types.W_Type(
            "INTEGER",
            [23, 21],
            interp_types.cast_integer
        )
        self.w_FLOAT = interp_types.W_Type(
            "FLOAT",
            [701, 700],
            interp_types.cast_float,
        )
        self.w_DECIMAL = interp_types.W_Type(
            "DECIMAL",
            [1700],
            interp_types.cast_decimal
        )
        self.w_BOOLEAN = interp_types.W_Type(
            "BOOLEAN",
            [16],
            interp_types.cast_boolean
        )
        self.w_BINARY = interp_types.W_Type(
            "BINARY",
            [17],
            interp_types.cast_binary
        )
        self.w_ROWID = interp_types.W_Type(
            "ROWID",
            [26],
            interp_types.cast_integer
        )
        self.w_INTEGERARRAY = interp_types.W_Type(
            "INTEGERARRAY",
            [1005, 1006, 1007],
            interp_types.cast_generic_array,
            base_caster=self.w_INTEGER
        )
        self.w_FLOATARRAY = interp_types.W_Type(
            "FLOATARRAY",
            [1017, 1021, 1022],
            interp_types.cast_generic_array,
            base_caster=self.w_FLOAT
        )
        self.w_DECIMALARRAY = interp_types.W_Type(
            "DECIMALARRAY",
            [1231],
            interp_types.cast_generic_array,
            base_caster=self.w_DECIMAL,
        )
        self.w_STRINGARRAY = interp_types.W_Type(
            "STRINGARRAY",
            [1002, 1003, 1009, 1014, 1015],
            interp_types.cast_generic_array,
            base_caster=self.w_STRING
        )
        self.w_BINARYARRAY = interp_types.W_Type(
            "BINARYARRAY",
            [1001],
            interp_types.cast_generic_array,
            base_caster=self.w_BINARY
        )
        self.w_DATETIME = interp_types.W_Type(
            "DATETIME",
            [1114, 1184, 704, 1186],
            interp_types.cast_datetime
        )
        self.w_DATE = interp_types.W_Type(
            "DATE",
            [1082],
            interp_types.cast_date
        )
        self.w_TIME = interp_types.W_Type(
            "TIME",
            [1083, 1266],
            interp_types.cast_time,
        )
        self.w_INTERVAL = interp_types.W_Type(
            "INTERVAL",
            [704, 1186],
            interp_types.cast_interval,
        )
        self.w_UNICODE = interp_types.W_Type(
            "UNICODE",
            [19, 18, 25, 1042, 1043],
            interp_types.cast_unicode
        )

        self.typecasts = {}
        self.default_typecast = self.w_STRING
        for caster in [
            "STRING", "NUMBER", "LONGINTEGER", "INTEGER", "FLOAT", "DECIMAL",
            "BOOLEAN", "BINARY", "ROWID", "INTEGERARRAY", "FLOATARRAY",
            "DECIMALARRAY", "STRINGARRAY", "BINARYARRAY", "DATETIME", "DATE",
            "TIME", "INTERVAL",
        ]:
            _register_type(space, getattr(self, "w_%s" % caster), self.typecasts)

        self.encodings = {
            "UNICODE": "utf_8",
            "UTF8": "utf_8",
        }

        self.w_DateTime = self.w_Date = self.w_Time = self.w_TimeDelta = self.w_Decimal = self.w_FixedOffsetTimezone = space.w_None

    def startup(self, space):
        self.w_DateTime, self.w_Date, self.w_Time, self.w_TimeDelta, self.w_Decimal, self.w_FixedOffsetTimezone = space.fixedview(space.appexec([], """():
             import datetime
             import decimal

             import psycopg2

             return datetime.datetime, datetime.date, datetime.time, datetime.timedelta, decimal.Decimal, psycopg2.tz.FixedOffsetTimezone
        """))

        self.adapters_w[self.w_DateTime] = space.gettypefor(W_DateTime)
        self.adapters_w[self.w_Date] = space.gettypefor(W_DateTime)
        self.adapters_w[self.w_Time] = space.gettypefor(W_DateTime)
        self.adapters_w[self.w_TimeDelta] = space.gettypefor(W_DateTime)
        self.adapters_w[self.w_Decimal] = space.gettypefor(W_Decimal)



def get(space):
    return space.fromcache(State)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.