-
:mod:`gzip` --- Support for :program:`gzip` files
=================================================
.. module:: gzip
:synopsis: Interfaces for gzip compression and decompression using file objects.
+This module provides a simple interface to compress and decompress files just
+like the GNU programs :program:`gzip` and :program:`gunzip` would.
+
+The data compression is provided by the :mod:``zlib`` module.
-The data compression provided by the ``zlib`` module is compatible with that
-used by the GNU compression program :program:`gzip`. Accordingly, the
-:mod:`gzip` module provides the :class:`GzipFile` class to read and write
+The :mod:`gzip` module provides the :class:`GzipFile` class which is modeled
+after Python's File Object. The :class:`GzipFile` class reads and writes
:program:`gzip`\ -format files, automatically compressing or decompressing the
-data so it looks like an ordinary file object. Note that additional file
-formats which can be decompressed by the :program:`gzip` and :program:`gunzip`
-programs, such as those produced by :program:`compress` and :program:`pack`,
-are not supported by this module.
+data so that it looks like an ordinary file object.
+
+Note that additional file formats which can be decompressed by the
+:program:`gzip` and :program:`gunzip` programs, such as those produced by
+:program:`compress` and :program:`pack`, are not supported by this module.
For other archive formats, see the :mod:`bz2`, :mod:`zipfile`, and
:mod:`tarfile` modules.
*compresslevel* defaults to ``9``.
+.. _gzip-usage-examples:
+
+Examples of usage
+-----------------
+
+Example of how to read a compressed file::
+
+ import gzip
+ f = gzip.open('/home/joe/file.txt.gz', 'rb')
+ file_content = f.read()
+ f.close()
+
+Example of how to create a compressed GZIP file::
+
+ import gzip
+ content = "Lots of content here"
+ f = gzip.open('/home/joe/file.txt.gz', 'wb')
+ f.write(content)
+ f.close()
+
+Example of how to GZIP compress an existing file::
+
+ import gzip
+ f_in = open('/home/joe/file.txt', 'rb')
+ f_out = gzip.open('/home/joe/file.txt.gz', 'wb')
+ f_out.writelines(f_in)
+ f_out.close()
+ f_in.close()
+
+
.. seealso::
Module :mod:`zlib`
deleted since the database connection was opened.
+.. attribute:: Connection.iterdump
+
+ Returns an iterator to dump the database in an SQL text format. Useful when
+ saving an in-memory database for later restoration. This function provides
+ the same capabilities as the :kbd:`.dump` command in the :program:`sqlite3`
+ shell.
+
+ .. versionadded:: 2.6
+
+ Example::
+
+ # Convert file existing_db.db to SQL dump file dump.sql
+ import sqlite3, os
+
+ con = sqlite3.connect('existing_db.db')
+ full_dump = os.linesep.join([line for line in con.iterdump()])
+ f = open('dump.sql', 'w')
+ f.writelines(full_dump)
+ f.close()
+
+
.. _sqlite3-cursor-objects:
Cursor Objects
#ifdef __CHAR_UNSIGNED__
#define Py_CHARMASK(c) (c)
#else
-#define Py_CHARMASK(c) ((c) & 0xff)
+#define Py_CHARMASK(c) ((unsigned char)((c) & 0xff))
#endif
#include "pyfpe.h"
extern const unsigned int _Py_ctype_table[256];
-#define ISLOWER(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_LOWER)
-#define ISUPPER(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_UPPER)
-#define ISALPHA(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_ALPHA)
-#define ISDIGIT(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_DIGIT)
-#define ISXDIGIT(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_XDIGIT)
-#define ISALNUM(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_ALNUM)
-#define ISSPACE(c) (_Py_ctype_table[(unsigned)Py_CHARMASK(c)] & FLAG_SPACE)
+#define ISLOWER(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_LOWER)
+#define ISUPPER(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_UPPER)
+#define ISALPHA(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_ALPHA)
+#define ISDIGIT(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_DIGIT)
+#define ISXDIGIT(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_XDIGIT)
+#define ISALNUM(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_ALNUM)
+#define ISSPACE(c) (_Py_ctype_table[Py_CHARMASK(c)] & FLAG_SPACE)
#undef islower
#define islower(c) undefined_islower(c)
'test_support',
'test_future1',
'test_future2',
- 'test_future3',
}
def findtests(testdir=None, stdtests=STDTESTS, nottests=NOTTESTS):
# ===== InputSource support
- def test_expat_inpsource_filename(self):
+ def XXXtest_expat_inpsource_filename(self):
+ # FIXME: test blocks indefinitely
parser = create_parser()
result = StringIO()
xmlgen = XMLGenerator(result)
self.assertEquals(result.getvalue(), xml_test_out)
- def test_expat_inpsource_sysid(self):
+ def XXXtest_expat_inpsource_sysid(self):
+ # FIXME: test blocks indefinitely
parser = create_parser()
result = StringIO()
xmlgen = XMLGenerator(result)
self.assertEquals(parser.getPublicId(), None)
self.assertEquals(parser.getLineNumber(), 1)
- def test_expat_locator_withinfo(self):
+ def XXXtest_expat_locator_withinfo(self):
+ # FIXME: test blocks indefinitely
result = StringIO()
xmlgen = XMLGenerator(result)
parser = create_parser()
self.assertRaises(SAXParseException, parser.parse, sio)
-def unittest_main():
+def test_main():
run_unittest(MakeParserTest,
SaxutilsTest,
XmlgenTest,
XmlReaderTest)
if __name__ == "__main__":
- unittest_main()
+ test_main()
import socket
import select
-import time
import thread, threading
+import time
+import traceback
import Queue
import sys
import os
except Alarm:
pass
except:
- self.fail("caught other exception instead of Alarm")
+ self.fail("caught other exception instead of Alarm:"
+ " %s(%s):\n%s" %
+ (sys.exc_info()[:2] + (traceback.format_exc(),)))
else:
self.fail("nothing caught")
- signal.alarm(0) # shut off alarm
+ finally:
+ signal.alarm(0) # shut off alarm
except Alarm:
self.fail("got Alarm in wrong place")
finally:
except ImportError:
raise TestSkipped('no sqlite available')
from sqlite3.test import (dbapi, types, userfunctions,
- factory, transactions, hooks, regression)
+ factory, transactions, hooks, regression,
+ dump)
def test_main():
run_unittest(dbapi.suite(), types.suite(), userfunctions.suite(),
- factory.suite(), transactions.suite(), hooks.suite(),
- regression.suite())
+ factory.suite(), transactions.suite(),
+ hooks.suite(), regression.suite(), dump.suite())
if __name__ == "__main__":
test_main()
from unittest import TestCase
from test import test_support
+PORT = 9091
def server(evt):
serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serv.settimeout(3)
serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- serv.bind(("", 9091))
+ global PORT
+ PORT = test_support.bind_port(serv, "", PORT)
serv.listen(5)
evt.set()
try:
def testBasic(self):
# connects
- telnet = telnetlib.Telnet("localhost", 9091)
+ telnet = telnetlib.Telnet("localhost", PORT)
telnet.sock.close()
def testTimeoutDefault(self):
# default
- telnet = telnetlib.Telnet("localhost", 9091)
+ telnet = telnetlib.Telnet("localhost", PORT)
self.assertTrue(telnet.sock.gettimeout() is None)
telnet.sock.close()
def testTimeoutValue(self):
# a value
- telnet = telnetlib.Telnet("localhost", 9091, timeout=30)
+ telnet = telnetlib.Telnet("localhost", PORT, timeout=30)
self.assertEqual(telnet.sock.gettimeout(), 30)
telnet.sock.close()
def testTimeoutDifferentOrder(self):
telnet = telnetlib.Telnet(timeout=30)
- telnet.open("localhost", 9091)
+ telnet.open("localhost", PORT)
self.assertEqual(telnet.sock.gettimeout(), 30)
telnet.sock.close()
previous = socket.getdefaulttimeout()
socket.setdefaulttimeout(30)
try:
- telnet = telnetlib.Telnet("localhost", 9091, timeout=None)
+ telnet = telnetlib.Telnet("localhost", PORT, timeout=None)
finally:
socket.setdefaulttimeout(previous)
self.assertEqual(telnet.sock.gettimeout(), 30)
weak_cyclic_object = weakref.ref(cyclic_object)
cyclic_object.thread.join()
del cyclic_object
- self.assertEquals(None, weak_cyclic_object())
+ self.assertEquals(None, weak_cyclic_object(),
+ msg=('%d references still around' %
+ sys.getrefcount(weak_cyclic_object())))
raising_cyclic_object = RunSelfFunction(should_raise=True)
weak_raising_cyclic_object = weakref.ref(raising_cyclic_object)
raising_cyclic_object.thread.join()
del raising_cyclic_object
- self.assertEquals(None, weak_raising_cyclic_object())
+ self.assertEquals(None, weak_raising_cyclic_object(),
+ msg=('%d references still around' %
+ sys.getrefcount(weak_raising_cyclic_object())))
class ThreadingExceptionTests(unittest.TestCase):
#!/usr/bin/env python
+import mimetools
import threading
import urlparse
import urllib2
# Test cases
class ProxyAuthTests(unittest.TestCase):
- URL = "http://www.foo.com"
+ URL = "http://localhost"
USER = "tester"
PASSWD = "test123"
pass
result.close()
+
+def GetRequestHandler(responses):
+
+ class FakeHTTPRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
+
+ server_version = "TestHTTP/"
+ requests = []
+ headers_received = []
+ port = 80
+
+ def do_GET(self):
+ body = self.send_head()
+ if body:
+ self.wfile.write(body)
+
+ def do_POST(self):
+ content_length = self.headers['Content-Length']
+ post_data = self.rfile.read(int(content_length))
+ self.do_GET()
+ self.requests.append(post_data)
+
+ def send_head(self):
+ FakeHTTPRequestHandler.headers_received = self.headers
+ self.requests.append(self.path)
+ response_code, headers, body = responses.pop(0)
+
+ self.send_response(response_code)
+
+ for (header, value) in headers:
+ self.send_header(header, value % self.port)
+ if body:
+ self.send_header('Content-type', 'text/plain')
+ self.end_headers()
+ return body
+ self.end_headers()
+
+ def log_message(self, *args):
+ pass
+
+
+ return FakeHTTPRequestHandler
+
+
+class TestUrlopen(unittest.TestCase):
+ """Tests urllib2.urlopen using the network.
+
+ These tests are not exhaustive. Assuming that testing using files does a
+ good job overall of some of the basic interface features. There are no
+ tests exercising the optional 'data' and 'proxies' arguments. No tests
+ for transparent redirection have been written.
+ """
+
+ def start_server(self, responses):
+ handler = GetRequestHandler(responses)
+
+ self.server = LoopbackHttpServerThread(handler)
+ self.server.start()
+ self.server.ready.wait()
+ port = self.server.port
+ handler.port = port
+ return handler
+
+
+ def test_redirection(self):
+ expected_response = b'We got here...'
+ responses = [
+ (302, [('Location', 'http://localhost:%s/somewhere_else')], ''),
+ (200, [], expected_response)
+ ]
+
+ handler = self.start_server(responses)
+
+ try:
+ f = urllib2.urlopen('http://localhost:%s/' % handler.port)
+ data = f.read()
+ f.close()
+
+ self.assertEquals(data, expected_response)
+ self.assertEquals(handler.requests, ['/', '/somewhere_else'])
+ finally:
+ self.server.stop()
+
+
+ def test_404(self):
+ expected_response = b'Bad bad bad...'
+ handler = self.start_server([(404, [], expected_response)])
+
+ try:
+ try:
+ urllib2.urlopen('http://localhost:%s/weeble' % handler.port)
+ except urllib2.URLError as f:
+ data = f.read()
+ f.close()
+ else:
+ self.fail('404 should raise URLError')
+
+ self.assertEquals(data, expected_response)
+ self.assertEquals(handler.requests, ['/weeble'])
+ finally:
+ self.server.stop()
+
+
+ def test_200(self):
+ expected_response = b'pycon 2008...'
+ handler = self.start_server([(200, [], expected_response)])
+
+ try:
+ f = urllib2.urlopen('http://localhost:%s/bizarre' % handler.port)
+ data = f.read()
+ f.close()
+
+ self.assertEquals(data, expected_response)
+ self.assertEquals(handler.requests, ['/bizarre'])
+ finally:
+ self.server.stop()
+
+ def test_200_with_parameters(self):
+ expected_response = b'pycon 2008...'
+ handler = self.start_server([(200, [], expected_response)])
+
+ try:
+ f = urllib2.urlopen('http://localhost:%s/bizarre' % handler.port, b'get=with_feeling')
+ data = f.read()
+ f.close()
+
+ self.assertEquals(data, expected_response)
+ self.assertEquals(handler.requests, ['/bizarre', b'get=with_feeling'])
+ finally:
+ self.server.stop()
+
+
+ def test_sending_headers(self):
+ handler = self.start_server([(200, [], b"we don't care")])
+
+ try:
+ req = urllib2.Request("http://localhost:%s/" % handler.port,
+ headers={'Range': 'bytes=20-39'})
+ urllib2.urlopen(req)
+ self.assertEqual(handler.headers_received['Range'], 'bytes=20-39')
+ finally:
+ self.server.stop()
+
+ def test_basic(self):
+ handler = self.start_server([(200, [], b"we don't care")])
+
+ try:
+ open_url = urllib2.urlopen("http://localhost:%s" % handler.port)
+ for attr in ("read", "close", "info", "geturl"):
+ self.assert_(hasattr(open_url, attr), "object returned from "
+ "urlopen lacks the %s attribute" % attr)
+ try:
+ self.assert_(open_url.read(), "calling 'read' failed")
+ finally:
+ open_url.close()
+ finally:
+ self.server.stop()
+
+ def test_info(self):
+ handler = self.start_server([(200, [], b"we don't care")])
+
+ try:
+ open_url = urllib2.urlopen("http://localhost:%s" % handler.port)
+ info_obj = open_url.info()
+ self.assert_(isinstance(info_obj, mimetools.Message),
+ "object returned by 'info' is not an instance of "
+ "mimetools.Message")
+ self.assertEqual(info_obj.getsubtype(), "plain")
+ finally:
+ self.server.stop()
+
+ def test_geturl(self):
+ # Make sure same URL as opened is returned by geturl.
+ handler = self.start_server([(200, [], b"we don't care")])
+
+ try:
+ open_url = urllib2.urlopen("http://localhost:%s" % handler.port)
+ url = open_url.geturl()
+ self.assertEqual(url, "http://localhost:%s" % handler.port)
+ finally:
+ self.server.stop()
+
+
+ def test_bad_address(self):
+ # Make sure proper exception is raised when connecting to a bogus
+ # address.
+ self.assertRaises(IOError,
+ # SF patch 809915: In Sep 2003, VeriSign started
+ # highjacking invalid .com and .net addresses to
+ # boost traffic to their own site. This test
+ # started failing then. One hopes the .invalid
+ # domain will be spared to serve its defined
+ # purpose.
+ # urllib2.urlopen, "http://www.sadflkjsasadf.com/")
+ urllib2.urlopen, "http://www.python.invalid./")
+
+
def test_main():
# We will NOT depend on the network resource flag
# (Lib/test/regrtest.py -u network) since all tests here are only
#test_support.requires("network")
test_support.run_unittest(ProxyAuthTests)
+ test_support.run_unittest(TestUrlopen)
if __name__ == "__main__":
test_main()
raise last_exc
-class URLTimeoutTest(unittest.TestCase):
-
- TIMEOUT = 10.0
-
- def setUp(self):
- socket.setdefaulttimeout(self.TIMEOUT)
-
- def tearDown(self):
- socket.setdefaulttimeout(None)
-
- def testURLread(self):
- f = _urlopen_with_retry("http://www.python.org/")
- x = f.read()
-
class AuthTests(unittest.TestCase):
"""Tests urllib2 authentication features."""
response.close()
self.assert_(fileobject.closed)
-class urlopenNetworkTests(unittest.TestCase):
- """Tests urllib2.urlopen using the network.
-
- These tests are not exhaustive. Assuming that testing using files does a
- good job overall of some of the basic interface features. There are no
- tests exercising the optional 'data' and 'proxies' arguments. No tests
- for transparent redirection have been written.
-
- setUp is not used for always constructing a connection to
- http://www.python.org/ since there a few tests that don't use that address
- and making a connection is expensive enough to warrant minimizing unneeded
- connections.
-
- """
-
- def test_basic(self):
- # Simple test expected to pass.
- open_url = _urlopen_with_retry("http://www.python.org/")
- for attr in ("read", "close", "info", "geturl"):
- self.assert_(hasattr(open_url, attr), "object returned from "
- "urlopen lacks the %s attribute" % attr)
- try:
- self.assert_(open_url.read(), "calling 'read' failed")
- finally:
- open_url.close()
-
- def test_info(self):
- # Test 'info'.
- open_url = _urlopen_with_retry("http://www.python.org/")
- try:
- info_obj = open_url.info()
- finally:
- open_url.close()
- self.assert_(isinstance(info_obj, mimetools.Message),
- "object returned by 'info' is not an instance of "
- "mimetools.Message")
- self.assertEqual(info_obj.getsubtype(), "html")
-
- def test_geturl(self):
- # Make sure same URL as opened is returned by geturl.
- URL = "http://www.python.org/"
- open_url = _urlopen_with_retry(URL)
- try:
- gotten_url = open_url.geturl()
- finally:
- open_url.close()
- self.assertEqual(gotten_url, URL)
-
- def test_bad_address(self):
- # Make sure proper exception is raised when connecting to a bogus
- # address.
- self.assertRaises(IOError,
- # SF patch 809915: In Sep 2003, VeriSign started
- # highjacking invalid .com and .net addresses to
- # boost traffic to their own site. This test
- # started failing then. One hopes the .invalid
- # domain will be spared to serve its defined
- # purpose.
- # urllib2.urlopen, "http://www.sadflkjsasadf.com/")
- urllib2.urlopen, "http://www.python.invalid./")
-
-
class OtherNetworkTests(unittest.TestCase):
def setUp(self):
if 0: # for debugging
logger = logging.getLogger("test_urllib2net")
logger.addHandler(logging.StreamHandler())
- def test_range (self):
- req = urllib2.Request("http://www.python.org",
- headers={'Range': 'bytes=20-39'})
- result = _urlopen_with_retry(req)
- data = result.read()
- self.assertEqual(len(data), 20)
-
# XXX The rest of these tests aren't very good -- they don't check much.
# They do sometimes catch some major disasters, though.
finally:
os.remove(TESTFN)
- def test_http(self):
- urls = [
- 'http://www.espn.com/', # redirect
- 'http://www.python.org/Spanish/Inquistion/',
- ('http://www.python.org/cgi-bin/faqw.py',
- 'query=pythonistas&querytype=simple&casefold=yes&req=search', None),
- 'http://www.python.org/',
- ]
- self._test_urls(urls, self._extra_handlers())
-
# XXX Following test depends on machine configurations that are internal
# to CNRI. Need to set up a public server with the right authentication
# configuration for test purposes.
return handlers
+
class TimeoutTest(unittest.TestCase):
def test_http_basic(self):
u = _urlopen_with_retry("http://www.python.org")
def test_main():
test_support.requires("network")
- test_support.run_unittest(URLTimeoutTest,
- urlopenNetworkTests,
- AuthTests,
+ test_support.run_unittest(AuthTests,
OtherNetworkTests,
CloseSocketTest,
TimeoutTest,
'''This is my function'''
return True
+ class MyXMLRPCServer(SimpleXMLRPCServer.SimpleXMLRPCServer):
+ def get_request(self):
+ # Ensure the socket is always non-blocking. On Linux, socket
+ # attributes are not inherited like they are on *BSD and Windows.
+ s, port = self.socket.accept()
+ s.setblocking(True)
+ return s, port
+
try:
- serv = SimpleXMLRPCServer.SimpleXMLRPCServer(("localhost", 0),
- logRequests=False, bind_and_activate=False)
+ serv = MyXMLRPCServer(("localhost", 0),
+ logRequests=False, bind_and_activate=False)
serv.server_bind()
global PORT
PORT = serv.socket.getsockname()[1]
# shutdown and thus raises an exception about trying to perform some
# operation on/with a NoneType
__exc_info = _sys.exc_info
+ # Keep sys.exc_clear too to clear the exception just before
+ # allowing .join() to return.
+ #XXX __exc_clear = _sys.exc_clear
def __init__(self, group=None, target=None, name=None,
args=(), kwargs=None, verbose=None):
else:
if __debug__:
self._note("%s._bootstrap(): normal return", self)
+ finally:
+ # Prevent a race in
+ # test_threading.test_no_refcycle_through_target when
+ # the exception keeps the target alive past when we
+ # assert that it's dead.
+ #XXX self.__exc_clear()
+ pass
finally:
with _active_limbo_lock:
self._stop()
Vivek Khera
Mads Kiilerich
Taek Joo Kim
+Paul Kippes
Steve Kirsch
Ron Klatchko
Bastian Kleineidam
/* connection.c - the connection type
*
- * Copyright (C) 2004-2006 Gerhard Häring <gh@ghaering.de>
+ * Copyright (C) 2004-2006 Gerhard H�ring <gh@ghaering.de>
*
* This file is part of pysqlite.
*
return retval;
}
+/* Function author: Paul Kippes <kippesp@gmail.com>
+ * Class method of Connection to call the Python function _iterdump
+ * of the sqlite3 module.
+ */
+static PyObject *
+pysqlite_connection_iterdump(pysqlite_Connection* self, PyObject* args)
+{
+ PyObject* retval = NULL;
+ PyObject* module = NULL;
+ PyObject* module_dict;
+ PyObject* pyfn_iterdump;
+
+ if (!pysqlite_check_connection(self)) {
+ goto finally;
+ }
+
+ module = PyImport_ImportModule(MODULE_NAME ".dump");
+ if (!module) {
+ goto finally;
+ }
+
+ module_dict = PyModule_GetDict(module);
+ if (!module_dict) {
+ goto finally;
+ }
+
+ pyfn_iterdump = PyDict_GetItemString(module_dict, "_iterdump");
+ if (!pyfn_iterdump) {
+ PyErr_SetString(pysqlite_OperationalError, "Failed to obtain _iterdump() reference");
+ goto finally;
+ }
+
+ args = PyTuple_New(1);
+ if (!args) {
+ goto finally;
+ }
+ Py_INCREF(self);
+ PyTuple_SetItem(args, 0, (PyObject*)self);
+ retval = PyObject_CallObject(pyfn_iterdump, args);
+
+finally:
+ Py_XDECREF(args);
+ Py_XDECREF(module);
+ return retval;
+}
+
static PyObject *
pysqlite_connection_create_collation(pysqlite_Connection* self, PyObject* args)
{
return retval;
}
+/* Called when the connection is used as a context manager. Returns itself as a
+ * convenience to the caller. */
+static PyObject *
+pysqlite_connection_enter(pysqlite_Connection* self, PyObject* args)
+{
+ Py_INCREF(self);
+ return (PyObject*)self;
+}
+
+/** Called when the connection is used as a context manager. If there was any
+ * exception, a rollback takes place; otherwise we commit. */
+static PyObject *
+pysqlite_connection_exit(pysqlite_Connection* self, PyObject* args)
+{
+ PyObject* exc_type, *exc_value, *exc_tb;
+ char* method_name;
+ PyObject* result;
+
+ if (!PyArg_ParseTuple(args, "OOO", &exc_type, &exc_value, &exc_tb)) {
+ return NULL;
+ }
+
+ if (exc_type == Py_None && exc_value == Py_None && exc_tb == Py_None) {
+ method_name = "commit";
+ } else {
+ method_name = "rollback";
+ }
+
+ result = PyObject_CallMethod((PyObject*)self, method_name, "");
+ if (!result) {
+ return NULL;
+ }
+ Py_DECREF(result);
+
+ Py_RETURN_FALSE;
+}
+
static char connection_doc[] =
PyDoc_STR("SQLite database connection object.");
PyDoc_STR("Creates a collation function. Non-standard.")},
{"interrupt", (PyCFunction)pysqlite_connection_interrupt, METH_NOARGS,
PyDoc_STR("Abort any pending database operation. Non-standard.")},
+ {"iterdump", (PyCFunction)pysqlite_connection_iterdump, METH_NOARGS,
+ PyDoc_STR("Returns iterator to the dump of the database in an SQL text"
+ "format.")},
+ {"__enter__", (PyCFunction)pysqlite_connection_enter, METH_NOARGS,
+ PyDoc_STR("For context manager. Non-standard.")},
+ {"__exit__", (PyCFunction)pysqlite_connection_exit, METH_VARARGS,
+ PyDoc_STR("For context manager. Non-standard.")},
{NULL, NULL}
};
#include "Python.h"
#include "structmember.h"
+#ifdef HAVE_IEEEFP_H
+#include <ieeefp.h>
+#endif
+
#ifndef WITHOUT_COMPLEX
/* Precisions used by repr() and str(), respectively.
#include <ctype.h>
#include <float.h>
+#ifdef HAVE_IEEEFP_H
+#include <ieeefp.h>
+#endif
+
#if !defined(__STDC__)
extern double fmod(double, double);
extern double pow(double, double);
n >>= 1;
/* n <- total # of bits needed, while setting p to end-of-string */
n = 0;
- while (_PyLong_DigitValue[(unsigned)Py_CHARMASK(*p)] < base)
+ while (_PyLong_DigitValue[Py_CHARMASK(*p)] < base)
++p;
*str = p;
/* n <- # of Python digits needed, = ceiling(n/PyLong_SHIFT). */
bits_in_accum = 0;
pdigit = z->ob_digit;
while (--p >= start) {
- int k = _PyLong_DigitValue[(unsigned)Py_CHARMASK(*p)];
+ int k = _PyLong_DigitValue[Py_CHARMASK(*p)];
assert(k >= 0 && k < base);
accum |= (twodigits)(k << bits_in_accum);
bits_in_accum += bits_per_char;
/* Find length of the string of numeric characters. */
scan = str;
- while (_PyLong_DigitValue[(unsigned)Py_CHARMASK(*scan)] < base)
+ while (_PyLong_DigitValue[Py_CHARMASK(*scan)] < base)
++scan;
/* Create a long object that can contain the largest possible
/* Work ;-) */
while (str < scan) {
/* grab up to convwidth digits from the input string */
- c = (digit)_PyLong_DigitValue[(unsigned)Py_CHARMASK(*str++)];
+ c = (digit)_PyLong_DigitValue[Py_CHARMASK(*str++)];
for (i = 1; i < convwidth && str != scan; ++i, ++str) {
c = (twodigits)(c * base +
- _PyLong_DigitValue[(unsigned)Py_CHARMASK(*str)]);
+ _PyLong_DigitValue[Py_CHARMASK(*str)]);
assert(c < PyLong_BASE);
}
/* Single characters are shared when using this constructor.
Restrict to ASCII, since the input must be UTF-8. */
if (size == 1 && Py_CHARMASK(*u) < 128) {
- unicode = unicode_latin1[(unsigned)Py_CHARMASK(*u)];
+ unicode = unicode_latin1[Py_CHARMASK(*u)];
if (!unicode) {
unicode = _PyUnicode_New(1);
if (!unicode)
return NULL;
unicode->str[0] = Py_CHARMASK(*u);
- unicode_latin1[(unsigned)Py_CHARMASK(*u)] = unicode;
+ unicode_latin1[Py_CHARMASK(*u)] = unicode;
}
Py_INCREF(unicode);
return (PyObject *)unicode;
/* Don't ever change this -- it would break the portability of Python code */
#define TABSIZE 8
-/* Convert a possibly signed character to a nonnegative int */
-/* XXX This assumes characters are 8 bits wide */
-#ifdef __CHAR_UNSIGNED__
-#define Py_CHARMASK(c) (c)
-#else
-#define Py_CHARMASK(c) ((c) & 0xff)
-#endif
-
/* Forward */
static struct tok_state *tok_new(void);
static int tok_nextc(struct tok_state *tok);
++str;
if (*str == 'x' || *str == 'X') {
/* there must be at least one digit after 0x */
- if (_PyLong_DigitValue[(unsigned)Py_CHARMASK(str[1])] >= 16) {
+ if (_PyLong_DigitValue[Py_CHARMASK(str[1])] >= 16) {
if (ptr)
*ptr = str;
return 0;
base = 16;
} else if (*str == 'o' || *str == 'O') {
/* there must be at least one digit after 0o */
- if (_PyLong_DigitValue[(unsigned)Py_CHARMASK(str[1])] >= 8) {
+ if (_PyLong_DigitValue[Py_CHARMASK(str[1])] >= 8) {
if (ptr)
*ptr = str;
return 0;
base = 8;
} else if (*str == 'b' || *str == 'B') {
/* there must be at least one digit after 0b */
- if (_PyLong_DigitValue[(unsigned)Py_CHARMASK(str[1])] >= 2) {
+ if (_PyLong_DigitValue[Py_CHARMASK(str[1])] >= 2) {
if (ptr)
*ptr = str;
return 0;
++str;
if (*str == 'x' || *str == 'X') {
/* there must be at least one digit after 0x */
- if (_PyLong_DigitValue[(unsigned)Py_CHARMASK(str[1])] >= 16) {
+ if (_PyLong_DigitValue[Py_CHARMASK(str[1])] >= 16) {
if (ptr)
*ptr = str;
return 0;
++str;
if (*str == 'o' || *str == 'O') {
/* there must be at least one digit after 0o */
- if (_PyLong_DigitValue[(unsigned)Py_CHARMASK(str[1])] >= 8) {
+ if (_PyLong_DigitValue[Py_CHARMASK(str[1])] >= 8) {
if (ptr)
*ptr = str;
return 0;
++str;
if (*str == 'b' || *str == 'B') {
/* there must be at least one digit after 0b */
- if (_PyLong_DigitValue[(unsigned)Py_CHARMASK(str[1])] >= 2) {
+ if (_PyLong_DigitValue[Py_CHARMASK(str[1])] >= 2) {
if (ptr)
*ptr = str;
return 0;
ovlimit = digitlimit[base];
/* do the conversion until non-digit character encountered */
- while ((c = _PyLong_DigitValue[(unsigned)Py_CHARMASK(*str)]) < base) {
+ while ((c = _PyLong_DigitValue[Py_CHARMASK(*str)]) < base) {
if (ovlimit > 0) /* no overflow check required */
result = result * base + c;
else { /* requires overflow check */
overflowed:
if (ptr) {
/* spool through remaining digit characters */
- while (_PyLong_DigitValue[(unsigned)Py_CHARMASK(*str)] < base)
+ while (_PyLong_DigitValue[Py_CHARMASK(*str)] < base)
++str;
*ptr = str;
}
#! /bin/sh
-# From configure.in Revision: 61728 .
+# From configure.in Revision: 61848 .
# Guess values for system-dependent variables and create Makefiles.
# Generated by GNU Autoconf 2.61 for python 3.0.
#
+
for ac_header in asm/types.h conio.h curses.h direct.h dlfcn.h errno.h \
fcntl.h grp.h \
-io.h langinfo.h libintl.h ncurses.h poll.h process.h pthread.h \
+ieeefp.h io.h langinfo.h libintl.h ncurses.h poll.h process.h pthread.h \
shadow.h signal.h stdint.h stropts.h termios.h thread.h \
unistd.h utime.h \
sys/audioio.h sys/bsdtty.h sys/epoll.h sys/event.h sys/file.h sys/loadavg.h \
AC_HEADER_STDC
AC_CHECK_HEADERS(asm/types.h conio.h curses.h direct.h dlfcn.h errno.h \
fcntl.h grp.h \
-io.h langinfo.h libintl.h ncurses.h poll.h process.h pthread.h \
+ieeefp.h io.h langinfo.h libintl.h ncurses.h poll.h process.h pthread.h \
shadow.h signal.h stdint.h stropts.h termios.h thread.h \
unistd.h utime.h \
sys/audioio.h sys/bsdtty.h sys/epoll.h sys/event.h sys/file.h sys/loadavg.h \
/* Define to 1 if you have the `hypot' function. */
#undef HAVE_HYPOT
+/* Define to 1 if you have the <ieeefp.h> header file. */
+#undef HAVE_IEEEFP_H
+
/* Define if you have the 'inet_aton' function. */
#undef HAVE_INET_ATON
projects / python / commitdiff