import unittest
from test.support.os_helper import TESTFN, unlink
+from test.support import threading_helper
# Helper for tests using TESTFN
with managed_connect(f"file:{TESTFN}?mode=ro", uri=True) as cx:
cx.execute(self._sql)
+ def test_database_keyword(self):
+ with sqlite.connect(database=":memory:") as cx:
+ self.assertEqual(type(cx), sqlite.Connection)
+
class CursorTests(unittest.TestCase):
def setUp(self):
if len(errors) > 0:
self.fail("\n".join(errors))
+ @threading_helper.reap_threads
+ def test_dont_check_same_thread(self):
+ def run(con, err):
+ try:
+ con.execute("select 1")
+ except sqlite.Error:
+ err.append("multi-threading not allowed")
+
+ con = sqlite.connect(":memory:", check_same_thread=False)
+ err = []
+ t = threading.Thread(target=run, kwargs={"con": con, "err": err})
+ t.start()
+ t.join()
+ self.assertEqual(len(err), 0, "\n".join(err))
+
+
class ConstructorTests(unittest.TestCase):
def test_date(self):
d = sqlite.Date(2004, 10, 28)
--- /dev/null
+Use Argument Clinic in :mod:`sqlite3`. Patches by Erlend E. Aasland.
preserve
[clinic start generated code]*/
+static int
+pysqlite_connection_init_impl(pysqlite_Connection *self,
+ PyObject *database_obj, double timeout,
+ int detect_types, PyObject *isolation_level,
+ int check_same_thread, PyObject *factory,
+ int cached_statements, int uri);
+
+static int
+pysqlite_connection_init(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+ int return_value = -1;
+ static const char * const _keywords[] = {"database", "timeout", "detect_types", "isolation_level", "check_same_thread", "factory", "cached_statements", "uri", NULL};
+ static _PyArg_Parser _parser = {NULL, _keywords, "Connection", 0};
+ PyObject *argsbuf[8];
+ PyObject * const *fastargs;
+ Py_ssize_t nargs = PyTuple_GET_SIZE(args);
+ Py_ssize_t noptargs = nargs + (kwargs ? PyDict_GET_SIZE(kwargs) : 0) - 1;
+ PyObject *database_obj;
+ double timeout = 5.0;
+ int detect_types = 0;
+ PyObject *isolation_level = NULL;
+ int check_same_thread = 1;
+ PyObject *factory = (PyObject*)clinic_state()->ConnectionType;
+ int cached_statements = 128;
+ int uri = 0;
+
+ fastargs = _PyArg_UnpackKeywords(_PyTuple_CAST(args)->ob_item, nargs, kwargs, NULL, &_parser, 1, 8, 0, argsbuf);
+ if (!fastargs) {
+ goto exit;
+ }
+ if (!PyUnicode_FSConverter(fastargs[0], &database_obj)) {
+ goto exit;
+ }
+ if (!noptargs) {
+ goto skip_optional_pos;
+ }
+ if (fastargs[1]) {
+ if (PyFloat_CheckExact(fastargs[1])) {
+ timeout = PyFloat_AS_DOUBLE(fastargs[1]);
+ }
+ else
+ {
+ timeout = PyFloat_AsDouble(fastargs[1]);
+ if (timeout == -1.0 && PyErr_Occurred()) {
+ goto exit;
+ }
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (fastargs[2]) {
+ detect_types = _PyLong_AsInt(fastargs[2]);
+ if (detect_types == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (fastargs[3]) {
+ isolation_level = fastargs[3];
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (fastargs[4]) {
+ check_same_thread = _PyLong_AsInt(fastargs[4]);
+ if (check_same_thread == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (fastargs[5]) {
+ factory = fastargs[5];
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (fastargs[6]) {
+ cached_statements = _PyLong_AsInt(fastargs[6]);
+ if (cached_statements == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ uri = PyObject_IsTrue(fastargs[7]);
+ if (uri < 0) {
+ goto exit;
+ }
+skip_optional_pos:
+ return_value = pysqlite_connection_init_impl((pysqlite_Connection *)self, database_obj, timeout, detect_types, isolation_level, check_same_thread, factory, cached_statements, uri);
+
+exit:
+ return return_value;
+}
+
PyDoc_STRVAR(pysqlite_connection_cursor__doc__,
"cursor($self, /, factory=<unrepresentable>)\n"
"--\n"
#ifndef PYSQLITE_CONNECTION_LOAD_EXTENSION_METHODDEF
#define PYSQLITE_CONNECTION_LOAD_EXTENSION_METHODDEF
#endif /* !defined(PYSQLITE_CONNECTION_LOAD_EXTENSION_METHODDEF) */
-/*[clinic end generated code: output=1ee2f6173f4acec3 input=a9049054013a1b77]*/
+/*[clinic end generated code: output=c350732a2758c8c1 input=a9049054013a1b77]*/
preserve
[clinic start generated code]*/
+PyDoc_STRVAR(pysqlite_connect__doc__,
+"connect($module, /, database, timeout=5.0, detect_types=0,\n"
+" isolation_level=<unrepresentable>, check_same_thread=True,\n"
+" factory=ConnectionType, cached_statements=128, uri=False)\n"
+"--\n"
+"\n"
+"Opens a connection to the SQLite database file database.\n"
+"\n"
+"You can use \":memory:\" to open a database connection to a database that resides\n"
+"in RAM instead of on disk.");
+
+#define PYSQLITE_CONNECT_METHODDEF \
+ {"connect", (PyCFunction)(void(*)(void))pysqlite_connect, METH_FASTCALL|METH_KEYWORDS, pysqlite_connect__doc__},
+
+static PyObject *
+pysqlite_connect_impl(PyObject *module, PyObject *database, double timeout,
+ int detect_types, PyObject *isolation_level,
+ int check_same_thread, PyObject *factory,
+ int cached_statements, int uri);
+
+static PyObject *
+pysqlite_connect(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
+{
+ PyObject *return_value = NULL;
+ static const char * const _keywords[] = {"database", "timeout", "detect_types", "isolation_level", "check_same_thread", "factory", "cached_statements", "uri", NULL};
+ static _PyArg_Parser _parser = {NULL, _keywords, "connect", 0};
+ PyObject *argsbuf[8];
+ Py_ssize_t noptargs = nargs + (kwnames ? PyTuple_GET_SIZE(kwnames) : 0) - 1;
+ PyObject *database;
+ double timeout = 5.0;
+ int detect_types = 0;
+ PyObject *isolation_level = NULL;
+ int check_same_thread = 1;
+ PyObject *factory = (PyObject*)clinic_state()->ConnectionType;
+ int cached_statements = 128;
+ int uri = 0;
+
+ args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 1, 8, 0, argsbuf);
+ if (!args) {
+ goto exit;
+ }
+ if (!PyUnicode_FSConverter(args[0], &database)) {
+ goto exit;
+ }
+ if (!noptargs) {
+ goto skip_optional_pos;
+ }
+ if (args[1]) {
+ if (PyFloat_CheckExact(args[1])) {
+ timeout = PyFloat_AS_DOUBLE(args[1]);
+ }
+ else
+ {
+ timeout = PyFloat_AsDouble(args[1]);
+ if (timeout == -1.0 && PyErr_Occurred()) {
+ goto exit;
+ }
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (args[2]) {
+ detect_types = _PyLong_AsInt(args[2]);
+ if (detect_types == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (args[3]) {
+ isolation_level = args[3];
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (args[4]) {
+ check_same_thread = _PyLong_AsInt(args[4]);
+ if (check_same_thread == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (args[5]) {
+ factory = args[5];
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ if (args[6]) {
+ cached_statements = _PyLong_AsInt(args[6]);
+ if (cached_statements == -1 && PyErr_Occurred()) {
+ goto exit;
+ }
+ if (!--noptargs) {
+ goto skip_optional_pos;
+ }
+ }
+ uri = PyObject_IsTrue(args[7]);
+ if (uri < 0) {
+ goto exit;
+ }
+skip_optional_pos:
+ return_value = pysqlite_connect_impl(module, database, timeout, detect_types, isolation_level, check_same_thread, factory, cached_statements, uri);
+
+exit:
+ return return_value;
+}
+
PyDoc_STRVAR(pysqlite_complete_statement__doc__,
"complete_statement($module, /, statement)\n"
"--\n"
exit:
return return_value;
}
-/*[clinic end generated code: output=e9c2442673289cab input=a9049054013a1b77]*/
+/*[clinic end generated code: output=ef03fdbf018d3391 input=a9049054013a1b77]*/
return res;
}
+/*[clinic input]
+_sqlite3.Connection.__init__ as pysqlite_connection_init
+
+ database as database_obj: object(converter='PyUnicode_FSConverter')
+ timeout: double = 5.0
+ detect_types: int = 0
+ isolation_level: object = NULL
+ check_same_thread: bool(accept={int}) = True
+ factory: object(c_default='(PyObject*)clinic_state()->ConnectionType') = ConnectionType
+ cached_statements: int = 128
+ uri: bool = False
+[clinic start generated code]*/
+
static int
-pysqlite_connection_init(pysqlite_Connection *self, PyObject *args,
- PyObject *kwargs)
+pysqlite_connection_init_impl(pysqlite_Connection *self,
+ PyObject *database_obj, double timeout,
+ int detect_types, PyObject *isolation_level,
+ int check_same_thread, PyObject *factory,
+ int cached_statements, int uri)
+/*[clinic end generated code: output=dc19df1c0e2b7b77 input=aa1f21bf12fe907a]*/
{
- static char *kwlist[] = {
- "database", "timeout", "detect_types", "isolation_level",
- "check_same_thread", "factory", "cached_statements", "uri",
- NULL
- };
-
- const char* database;
- PyObject* database_obj;
- int detect_types = 0;
- PyObject* isolation_level = NULL;
- PyObject* factory = NULL;
- int check_same_thread = 1;
- int cached_statements = 128;
- int uri = 0;
- double timeout = 5.0;
int rc;
- if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O&|diOiOip", kwlist,
- PyUnicode_FSConverter, &database_obj, &timeout, &detect_types,
- &isolation_level, &check_same_thread,
- &factory, &cached_statements, &uri))
- {
- return -1;
- }
-
if (PySys_Audit("sqlite3.connect", "O", database_obj) < 0) {
return -1;
}
- database = PyBytes_AsString(database_obj);
+ const char *database = PyBytes_AsString(database_obj);
self->initialized = 1;
(uri ? SQLITE_OPEN_URI : 0), NULL);
Py_END_ALLOW_THREADS
- Py_DECREF(database_obj);
+ Py_DECREF(database_obj); // needed bco. the AC FSConverter
if (rc != SQLITE_OK) {
_pysqlite_seterror(self->db);
pysqlite_state pysqlite_global_state;
-static PyObject* module_connect(PyObject* self, PyObject* args, PyObject*
- kwargs)
-{
- /* Python seems to have no way of extracting a single keyword-arg at
- * C-level, so this code is redundant with the one in connection_init in
- * connection.c and must always be copied from there ... */
+// NOTE: This must equal sqlite3.Connection.__init__ argument spec!
+/*[clinic input]
+_sqlite3.connect as pysqlite_connect
- static char *kwlist[] = {
- "database", "timeout", "detect_types", "isolation_level",
- "check_same_thread", "factory", "cached_statements", "uri",
- NULL
- };
- PyObject* database;
- int detect_types = 0;
- PyObject* isolation_level;
- PyObject* factory = NULL;
- int check_same_thread = 1;
- int cached_statements;
- int uri = 0;
- double timeout = 5.0;
-
- if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|diOiOip", kwlist,
- &database, &timeout, &detect_types,
- &isolation_level, &check_same_thread,
- &factory, &cached_statements, &uri))
- {
- return NULL;
- }
+ database: object(converter='PyUnicode_FSConverter')
+ timeout: double = 5.0
+ detect_types: int = 0
+ isolation_level: object = NULL
+ check_same_thread: bool(accept={int}) = True
+ factory: object(c_default='(PyObject*)clinic_state()->ConnectionType') = ConnectionType
+ cached_statements: int = 128
+ uri: bool = False
- if (factory == NULL) {
- pysqlite_state *state = pysqlite_get_state(self);
- factory = (PyObject *)state->ConnectionType;
- }
+Opens a connection to the SQLite database file database.
- return PyObject_Call(factory, args, kwargs);
-}
+You can use ":memory:" to open a database connection to a database that resides
+in RAM instead of on disk.
+[clinic start generated code]*/
-PyDoc_STRVAR(module_connect_doc,
-"connect(database[, timeout, detect_types, isolation_level,\n\
- check_same_thread, factory, cached_statements, uri])\n\
-\n\
-Opens a connection to the SQLite database file *database*. You can use\n\
-\":memory:\" to open a database connection to a database that resides in\n\
-RAM instead of on disk.");
+static PyObject *
+pysqlite_connect_impl(PyObject *module, PyObject *database, double timeout,
+ int detect_types, PyObject *isolation_level,
+ int check_same_thread, PyObject *factory,
+ int cached_statements, int uri)
+/*[clinic end generated code: output=450ac9078b4868bb input=ea6355ba55a78e12]*/
+{
+ if (isolation_level == NULL) {
+ isolation_level = PyUnicode_FromString("");
+ if (isolation_level == NULL) {
+ return NULL;
+ }
+ }
+ else {
+ Py_INCREF(isolation_level);
+ }
+ PyObject *res = PyObject_CallFunction(factory, "OdiOiOii", database,
+ timeout, detect_types,
+ isolation_level, check_same_thread,
+ factory, cached_statements, uri);
+ Py_DECREF(database); // needed bco. the AC FSConverter
+ Py_DECREF(isolation_level);
+ return res;
+}
/*[clinic input]
_sqlite3.complete_statement as pysqlite_complete_statement
}
static PyMethodDef module_methods[] = {
- {"connect", (PyCFunction)(void(*)(void))module_connect,
- METH_VARARGS | METH_KEYWORDS, module_connect_doc},
PYSQLITE_ADAPT_METHODDEF
PYSQLITE_COMPLETE_STATEMENT_METHODDEF
+ PYSQLITE_CONNECT_METHODDEF
PYSQLITE_ENABLE_CALLBACK_TRACE_METHODDEF
PYSQLITE_ENABLE_SHARED_CACHE_METHODDEF
PYSQLITE_REGISTER_ADAPTER_METHODDEF
projects / thirdparty / Python / cpython.git / commitdiff
