diff --git a/Lib/sqlite3/test/hooks.py b/Lib/sqlite3/test/hooks.py
index b798e74..2bfae10 100644
--- a/Lib/sqlite3/test/hooks.py
+++ b/Lib/sqlite3/test/hooks.py
@@ -76,6 +76,25 @@
         except sqlite.OperationalError, e:
             self.assertEqual(e.args[0].lower(), "no such collation sequence: mycoll")
 
+    def CheckCollationReturnsLargeInteger(self):
+        def mycoll(x, y):
+            # reverse order
+            return -((x > y) - (x < y)) * 2**32
+        con = sqlite.connect(":memory:")
+        con.create_collation("mycoll", mycoll)
+        sql = """
+            select x from (
+            select 'a' as x
+            union
+            select 'b' as x
+            union
+            select 'c' as x
+            ) order by x collate mycoll
+            """
+        result = con.execute(sql).fetchall()
+        self.assertEqual(result, [('c',), ('b',), ('a',)],
+                         msg="the expected order was not returned")
+
     def CheckCollationRegisterTwice(self):
         """
         Register two different collation functions under the same name.
diff --git a/Lib/sqlite3/test/userfunctions.py b/Lib/sqlite3/test/userfunctions.py
index 2db3a61..634812d 100644
--- a/Lib/sqlite3/test/userfunctions.py
+++ b/Lib/sqlite3/test/userfunctions.py
@@ -374,14 +374,15 @@
         val = cur.fetchone()[0]
         self.assertEqual(val, 60)
 
-def authorizer_cb(action, arg1, arg2, dbname, source):
-    if action != sqlite.SQLITE_SELECT:
-        return sqlite.SQLITE_DENY
-    if arg2 == 'c2' or arg1 == 't2':
-        return sqlite.SQLITE_DENY
-    return sqlite.SQLITE_OK
-
 class AuthorizerTests(unittest.TestCase):
+    @staticmethod
+    def authorizer_cb(action, arg1, arg2, dbname, source):
+        if action != sqlite.SQLITE_SELECT:
+            return sqlite.SQLITE_DENY
+        if arg2 == 'c2' or arg1 == 't2':
+            return sqlite.SQLITE_DENY
+        return sqlite.SQLITE_OK
+
     def setUp(self):
         self.con = sqlite.connect(":memory:")
         self.con.executescript("""
@@ -394,12 +395,12 @@
         # For our security test:
         self.con.execute("select c2 from t2")
 
-        self.con.set_authorizer(authorizer_cb)
+        self.con.set_authorizer(self.authorizer_cb)
 
     def tearDown(self):
         pass
 
-    def CheckTableAccess(self):
+    def test_table_access(self):
         try:
             self.con.execute("select * from t2")
         except sqlite.DatabaseError, e:
@@ -408,7 +409,7 @@
             return
         self.fail("should have raised an exception due to missing privileges")
 
-    def CheckColumnAccess(self):
+    def test_column_access(self):
         try:
             self.con.execute("select c2 from t1")
         except sqlite.DatabaseError, e:
@@ -417,11 +418,46 @@
             return
         self.fail("should have raised an exception due to missing privileges")
 
+class AuthorizerRaiseExceptionTests(AuthorizerTests):
+    @staticmethod
+    def authorizer_cb(action, arg1, arg2, dbname, source):
+        if action != sqlite.SQLITE_SELECT:
+            raise ValueError
+        if arg2 == 'c2' or arg1 == 't2':
+            raise ValueError
+        return sqlite.SQLITE_OK
+
+class AuthorizerIllegalTypeTests(AuthorizerTests):
+    @staticmethod
+    def authorizer_cb(action, arg1, arg2, dbname, source):
+        if action != sqlite.SQLITE_SELECT:
+            return 0.0
+        if arg2 == 'c2' or arg1 == 't2':
+            return 0.0
+        return sqlite.SQLITE_OK
+
+class AuthorizerLargeIntegerTests(AuthorizerTests):
+    @staticmethod
+    def authorizer_cb(action, arg1, arg2, dbname, source):
+        if action != sqlite.SQLITE_SELECT:
+            return 2**32
+        if arg2 == 'c2' or arg1 == 't2':
+            return 2**32
+        return sqlite.SQLITE_OK
+
+
 def suite():
     function_suite = unittest.makeSuite(FunctionTests, "Check")
     aggregate_suite = unittest.makeSuite(AggregateTests, "Check")
-    authorizer_suite = unittest.makeSuite(AuthorizerTests, "Check")
-    return unittest.TestSuite((function_suite, aggregate_suite, authorizer_suite))
+    authorizer_suite = unittest.makeSuite(AuthorizerTests)
+    return unittest.TestSuite((
+            function_suite,
+            aggregate_suite,
+            authorizer_suite,
+            unittest.makeSuite(AuthorizerRaiseExceptionTests),
+            unittest.makeSuite(AuthorizerIllegalTypeTests),
+            unittest.makeSuite(AuthorizerLargeIntegerTests),
+        ))
 
 def test():
     runner = unittest.TextTestRunner()
diff --git a/Misc/NEWS b/Misc/NEWS
index d1a57fd..4734212 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -202,6 +202,8 @@
 Library
 -------
 
+- Issue #17073: Fix some integer overflows in sqlite3 module.
+
 - Issue #6083: Fix multiple segmentation faults occured when PyArg_ParseTuple
   parses nested mutating sequence.
 
diff --git a/Modules/_sqlite/connection.c b/Modules/_sqlite/connection.c
index a646513..d89b4c1 100644
--- a/Modules/_sqlite/connection.c
+++ b/Modules/_sqlite/connection.c
@@ -538,39 +538,40 @@
     }
 }
 
-void _pysqlite_set_result(sqlite3_context* context, PyObject* py_val)
+static int
+_pysqlite_set_result(sqlite3_context* context, PyObject* py_val)
 {
-    const char* buffer;
-    Py_ssize_t buflen;
-    PyObject* stringval;
-
-    if ((!py_val) || PyErr_Occurred()) {
-        sqlite3_result_null(context);
-    } else if (py_val == Py_None) {
+    if (py_val == Py_None) {
         sqlite3_result_null(context);
     } else if (PyInt_Check(py_val)) {
         sqlite3_result_int64(context, (sqlite_int64)PyInt_AsLong(py_val));
     } else if (PyLong_Check(py_val)) {
-        sqlite3_result_int64(context, PyLong_AsLongLong(py_val));
+        sqlite_int64 value = _pysqlite_long_as_int64(py_val);
+        if (value == -1 && PyErr_Occurred())
+            return -1;
+        sqlite3_result_int64(context, value);
     } else if (PyFloat_Check(py_val)) {
         sqlite3_result_double(context, PyFloat_AsDouble(py_val));
     } else if (PyBuffer_Check(py_val)) {
+        const char* buffer;
+        Py_ssize_t buflen;
         if (PyObject_AsCharBuffer(py_val, &buffer, &buflen) != 0) {
             PyErr_SetString(PyExc_ValueError, "could not convert BLOB to buffer");
-        } else {
-            sqlite3_result_blob(context, buffer, buflen, SQLITE_TRANSIENT);
+            return -1;
         }
+        sqlite3_result_blob(context, buffer, buflen, SQLITE_TRANSIENT);
     } else if (PyString_Check(py_val)) {
         sqlite3_result_text(context, PyString_AsString(py_val), -1, SQLITE_TRANSIENT);
     } else if (PyUnicode_Check(py_val)) {
-        stringval = PyUnicode_AsUTF8String(py_val);
-        if (stringval) {
-            sqlite3_result_text(context, PyString_AsString(stringval), -1, SQLITE_TRANSIENT);
-            Py_DECREF(stringval);
-        }
+        PyObject * stringval = PyUnicode_AsUTF8String(py_val);
+        if (!stringval)
+            return -1;
+        sqlite3_result_text(context, PyString_AsString(stringval), -1, SQLITE_TRANSIENT);
+        Py_DECREF(stringval);
     } else {
-        /* TODO: raise error */
+        return -1;
     }
+    return 0;
 }
 
 PyObject* _pysqlite_build_py_params(sqlite3_context *context, int argc, sqlite3_value** argv)
@@ -580,7 +581,6 @@
     sqlite3_value* cur_value;
     PyObject* cur_py_value;
     const char* val_str;
-    sqlite_int64 val_int;
     Py_ssize_t buflen;
     void* raw_buffer;
 
@@ -593,11 +593,7 @@
         cur_value = argv[i];
         switch (sqlite3_value_type(argv[i])) {
             case SQLITE_INTEGER:
-                val_int = sqlite3_value_int64(cur_value);
-                if(val_int < LONG_MIN || val_int > LONG_MAX)
-                    cur_py_value = PyLong_FromLongLong(val_int);
-                else
-                    cur_py_value = PyInt_FromLong((long)val_int);
+                cur_py_value = _pysqlite_long_from_int64(sqlite3_value_int64(cur_value));
                 break;
             case SQLITE_FLOAT:
                 cur_py_value = PyFloat_FromDouble(sqlite3_value_double(cur_value));
@@ -648,6 +644,7 @@
     PyObject* args;
     PyObject* py_func;
     PyObject* py_retval = NULL;
+    int ok;
 
 #ifdef WITH_THREAD
     PyGILState_STATE threadstate;
@@ -663,10 +660,12 @@
         Py_DECREF(args);
     }
 
+    ok = 0;
     if (py_retval) {
-        _pysqlite_set_result(context, py_retval);
+        ok = _pysqlite_set_result(context, py_retval) == 0;
         Py_DECREF(py_retval);
-    } else {
+    }
+    if (!ok) {
         if (_enable_callback_tracebacks) {
             PyErr_Print();
         } else {
@@ -746,8 +745,9 @@
 
 void _pysqlite_final_callback(sqlite3_context* context)
 {
-    PyObject* function_result = NULL;
+    PyObject* function_result;
     PyObject** aggregate_instance;
+    int ok;
 
 #ifdef WITH_THREAD
     PyGILState_STATE threadstate;
@@ -764,21 +764,23 @@
     }
 
     function_result = PyObject_CallMethod(*aggregate_instance, "finalize", "");
-    if (!function_result) {
+    Py_DECREF(*aggregate_instance);
+
+    ok = 0;
+    if (function_result) {
+        ok = _pysqlite_set_result(context, function_result) == 0;
+        Py_DECREF(function_result);
+    }
+    if (!ok) {
         if (_enable_callback_tracebacks) {
             PyErr_Print();
         } else {
             PyErr_Clear();
         }
         _sqlite3_result_error(context, "user-defined aggregate's 'finalize' method raised error", -1);
-    } else {
-        _pysqlite_set_result(context, function_result);
     }
 
 error:
-    Py_XDECREF(*aggregate_instance);
-    Py_XDECREF(function_result);
-
 #ifdef WITH_THREAD
     PyGILState_Release(threadstate);
 #endif
@@ -935,7 +937,9 @@
         rc = SQLITE_DENY;
     } else {
         if (PyInt_Check(ret)) {
-            rc = (int)PyInt_AsLong(ret);
+            rc = _PyInt_AsInt(ret);
+            if (rc == -1 && PyErr_Occurred())
+                rc = SQLITE_DENY;
         } else {
             rc = SQLITE_DENY;
         }
@@ -967,7 +971,7 @@
         }
 
         /* abort query if error occurred */
-        rc = 1; 
+        rc = 1;
     } else {
         rc = (int)PyObject_IsTrue(ret);
         Py_DECREF(ret);
@@ -1337,6 +1341,7 @@
     PyGILState_STATE gilstate;
 #endif
     PyObject* retval = NULL;
+    long longval;
     int result = 0;
 #ifdef WITH_THREAD
     gilstate = PyGILState_Ensure();
@@ -1360,10 +1365,17 @@
         goto finally;
     }
 
-    result = PyInt_AsLong(retval);
-    if (PyErr_Occurred()) {
+    longval = PyLong_AsLongAndOverflow(retval, &result);
+    if (longval == -1 && PyErr_Occurred()) {
+        PyErr_Clear();
         result = 0;
     }
+    else if (!result) {
+        if (longval > 0)
+            result = 1;
+        else if (longval < 0)
+            result = -1;
+    }
 
 finally:
     Py_XDECREF(string1);
diff --git a/Modules/_sqlite/cursor.c b/Modules/_sqlite/cursor.c
index e1cd536..ae810de 100644
--- a/Modules/_sqlite/cursor.c
+++ b/Modules/_sqlite/cursor.c
@@ -26,14 +26,6 @@
 #include "util.h"
 #include "sqlitecompat.h"
 
-/* used to decide wether to call PyInt_FromLong or PyLong_FromLongLong */
-#ifndef INT32_MIN
-#define INT32_MIN (-2147483647 - 1)
-#endif
-#ifndef INT32_MAX
-#define INT32_MAX 2147483647
-#endif
-
 PyObject* pysqlite_cursor_iternext(pysqlite_Cursor* self);
 
 static char* errmsg_fetch_across_rollback = "Cursor needed to be reset because of commit/rollback and can no longer be fetched from.";
@@ -307,7 +299,6 @@
     PyObject* row;
     PyObject* item = NULL;
     int coltype;
-    PY_LONG_LONG intval;
     PyObject* converter;
     PyObject* converted;
     Py_ssize_t nbytes;
@@ -366,12 +357,7 @@
                 Py_INCREF(Py_None);
                 converted = Py_None;
             } else if (coltype == SQLITE_INTEGER) {
-                intval = sqlite3_column_int64(self->statement->st, i);
-                if (intval < INT32_MIN || intval > INT32_MAX) {
-                    converted = PyLong_FromLongLong(intval);
-                } else {
-                    converted = PyInt_FromLong((long)intval);
-                }
+                converted = _pysqlite_long_from_int64(sqlite3_column_int64(self->statement->st, i));
             } else if (coltype == SQLITE_FLOAT) {
                 converted = PyFloat_FromDouble(sqlite3_column_double(self->statement->st, i));
             } else if (coltype == SQLITE_TEXT) {
@@ -466,7 +452,6 @@
     PyObject* func_args;
     PyObject* result;
     int numcols;
-    PY_LONG_LONG lastrowid;
     int statement_type;
     PyObject* descriptor;
     PyObject* second_argument = NULL;
@@ -747,10 +732,11 @@
 
         Py_DECREF(self->lastrowid);
         if (!multiple && statement_type == STATEMENT_INSERT) {
+            sqlite3_int64 lastrowid;
             Py_BEGIN_ALLOW_THREADS
             lastrowid = sqlite3_last_insert_rowid(self->connection->db);
             Py_END_ALLOW_THREADS
-            self->lastrowid = PyInt_FromLong((long)lastrowid);
+            self->lastrowid = _pysqlite_long_from_int64(lastrowid);
         } else {
             Py_INCREF(Py_None);
             self->lastrowid = Py_None;
diff --git a/Modules/_sqlite/statement.c b/Modules/_sqlite/statement.c
index c777211..7a7a60f 100644
--- a/Modules/_sqlite/statement.c
+++ b/Modules/_sqlite/statement.c
@@ -26,6 +26,7 @@
 #include "connection.h"
 #include "microprotocols.h"
 #include "prepare_protocol.h"
+#include "util.h"
 #include "sqlitecompat.h"
 
 /* prototypes */
@@ -101,8 +102,6 @@
 int pysqlite_statement_bind_parameter(pysqlite_Statement* self, int pos, PyObject* parameter, int allow_8bit_chars)
 {
     int rc = SQLITE_OK;
-    long longval;
-    PY_LONG_LONG longlongval;
     const char* buffer;
     char* string;
     Py_ssize_t buflen;
@@ -153,15 +152,19 @@
     }
 
     switch (paramtype) {
-        case TYPE_INT:
-            longval = PyInt_AsLong(parameter);
-            rc = sqlite3_bind_int64(self->st, pos, (sqlite_int64)longval);
+        case TYPE_INT: {
+            long longval = PyInt_AsLong(parameter);
+            rc = sqlite3_bind_int64(self->st, pos, longval);
             break;
-        case TYPE_LONG:
-            longlongval = PyLong_AsLongLong(parameter);
-            /* in the overflow error case, longlongval is -1, and an exception is set */
-            rc = sqlite3_bind_int64(self->st, pos, (sqlite_int64)longlongval);
+        }
+        case TYPE_LONG: {
+            sqlite_int64 value = _pysqlite_long_as_int64(parameter);
+            if (value == -1 && PyErr_Occurred())
+                rc = -1;
+            else
+                rc = sqlite3_bind_int64(self->st, pos, (sqlite_int64)value);
             break;
+        }
         case TYPE_FLOAT:
             rc = sqlite3_bind_double(self->st, pos, PyFloat_AsDouble(parameter));
             break;
@@ -198,7 +201,7 @@
         return 1;
     }
 
-    if (PyInt_CheckExact(obj) || PyLong_CheckExact(obj) 
+    if (PyInt_CheckExact(obj) || PyLong_CheckExact(obj)
             || PyFloat_CheckExact(obj) || PyString_CheckExact(obj)
             || PyUnicode_CheckExact(obj) || PyBuffer_Check(obj)) {
         return 0;
diff --git a/Modules/_sqlite/util.c b/Modules/_sqlite/util.c
index 6b57b76..5f06b66 100644
--- a/Modules/_sqlite/util.c
+++ b/Modules/_sqlite/util.c
@@ -104,3 +104,69 @@
     return errorcode;
 }
 
+#ifdef WORDS_BIGENDIAN
+# define IS_LITTLE_ENDIAN 0
+#else
+# define IS_LITTLE_ENDIAN 1
+#endif
+
+PyObject *
+_pysqlite_long_from_int64(sqlite3_int64 value)
+{
+#ifdef HAVE_LONG_LONG
+# if SIZEOF_LONG_LONG < 8
+    if (value > PY_LLONG_MAX || value < PY_LLONG_MIN) {
+        return _PyLong_FromByteArray(&value, sizeof(value),
+                                     IS_LITTLE_ENDIAN, 1 /* signed */);
+    }
+# endif
+# if SIZEOF_LONG < SIZEOF_LONG_LONG
+    if (value > LONG_MAX || value < LONG_MIN)
+        return PyLong_FromLongLong(value);
+# endif
+#else
+# if SIZEOF_LONG < 8
+    if (value > LONG_MAX || value < LONG_MIN) {
+        return _PyLong_FromByteArray(&value, sizeof(value),
+                                     IS_LITTLE_ENDIAN, 1 /* signed */);
+    }
+# endif
+#endif
+    return PyInt_FromLong(value);
+}
+
+sqlite3_int64
+_pysqlite_long_as_int64(PyObject * py_val)
+{
+    int overflow;
+#ifdef HAVE_LONG_LONG
+    PY_LONG_LONG value = PyLong_AsLongLongAndOverflow(py_val, &overflow);
+#else
+    long value = PyLong_AsLongAndOverflow(py_val, &overflow);
+#endif
+    if (value == -1 && PyErr_Occurred())
+        return -1;
+    if (!overflow) {
+#ifdef HAVE_LONG_LONG
+# if SIZEOF_LONG_LONG > 8
+        if (-0x8000000000000000LL <= value && value <= 0x7FFFFFFFFFFFFFFFLL)
+# endif
+#else
+# if SIZEOF_LONG > 8
+        if (-0x8000000000000000L <= value && value <= 0x7FFFFFFFFFFFFFFFL)
+# endif
+#endif
+            return value;
+    }
+    else if (sizeof(value) < sizeof(sqlite3_int64)) {
+        sqlite3_int64 int64val;
+        if (_PyLong_AsByteArray((PyLongObject *)py_val,
+                                (unsigned char *)&int64val, sizeof(int64val),
+                                IS_LITTLE_ENDIAN, 1 /* signed */) >= 0) {
+            return int64val;
+        }
+    }
+    PyErr_SetString(PyExc_OverflowError,
+                    "Python int too large to convert to SQLite INTEGER");
+    return -1;
+}
diff --git a/Modules/_sqlite/util.h b/Modules/_sqlite/util.h
index 4269003..12a5710 100644
--- a/Modules/_sqlite/util.h
+++ b/Modules/_sqlite/util.h
@@ -35,4 +35,8 @@
  * Returns the error code (0 means no error occurred).
  */
 int _pysqlite_seterror(sqlite3* db, sqlite3_stmt* st);
+
+PyObject * _pysqlite_long_from_int64(sqlite3_int64 value);
+sqlite3_int64 _pysqlite_long_as_int64(PyObject * value);
+
 #endif