How to use pymssql - 10 common examples

def test_gevent_socket_pymssql_callproc_wait_read_concurrency(self):
        def wait_callback(read_fileno):
            gevent.socket.wait_read(read_fileno)

        pymssql.set_wait_callback(wait_callback)

        with pymssqlconn() as conn:
            cur = conn.cursor()
            proc_name = 'my_proc'
            cur.execute("IF OBJECT_ID('%s', 'P') IS NOT NULL DROP PROCEDURE %s" % (proc_name, proc_name))
            cur.execute("""
            CREATE PROCEDURE %s AS
            BEGIN
                SET NOCOUNT ON
                WAITFOR DELAY '00:00:05'  -- sleep for 5 seconds
                SELECT CURRENT_TIMESTAMP
            END
            """ % (proc_name,))
            conn.commit()

        elapsed_time = self._run_all_greenlets(

def test_tsql_to_python_exception_translation(self):
        """An error raised by a SP is translated to a PEP-249-dictated, pymssql layer exception."""
        # See https://github.com/pymssql/pymssql/issues/61
        cursor = self.pymssql.cursor()
        # Must raise an exception
        self.assertRaises(Exception, cursor.callproc, 'SPThatRaisesAnError')
        # Must be a PEP-249 exception, not a _mssql-layer one
        try:
            cursor.callproc('SPThatRaisesAnError')
        except Exception as e:
            self.assertTrue(isinstance(e,  pymssql.Error))
        # Must be a DatabaseError exception
        try:
            cursor.callproc('SPThatRaisesAnError')
        except Exception as e:
            self.assertTrue(isinstance(e,  pymssql.DatabaseError))

def test_version(self):
        assert pym.__version__

def test_tsql_to_python_exception_translation(self):
        """An error raised by a SP is translated to a PEP-249-dictated, pymssql layer exception."""
        # See https://github.com/pymssql/pymssql/issues/61
        cursor = self.pymssql.cursor()
        # Must raise an exception
        self.assertRaises(Exception, cursor.callproc, 'SPThatRaisesAnError')
        # Must be a PEP-249 exception, not a _mssql-layer one
        try:
            cursor.callproc('SPThatRaisesAnError')
        except Exception as e:
            self.assertTrue(isinstance(e,  pymssql.Error))
        # Must be a DatabaseError exception
        try:
            cursor.callproc('SPThatRaisesAnError')
        except Exception as e:
            self.assertTrue(isinstance(e,  pymssql.DatabaseError))

def test_rollback_after_create_error(self):
        """
            test_rollback_after_create_error

            For some reason, SQL server will issue a batch-abort
            if the statement is a CREATE statement and it fails.  This means
            the transaction is implicitly rolled back and a subsequent call to
            rollback() without special handling would result in an error.
        """
        cur = self.conn.cursor()
        cur.execute('insert into users values (%s)', 'foobar')
        eq_(self.row_count(), 1)
        try:
            cur.execute("CREATE TABLE badschema.t1 ( test1 CHAR(5) NOT NULL)")
        except pym.ProgrammingError as e:
            if 'badschema' not in str(e):
                raise
            # encountered an error, so we want to rollback
            self.conn.rollback()
        # rollback should have resulted in user's insert getting rolled back
        # too
        eq_(self.row_count(), 0)

def test_pymssql_Connection_with(self):
        with pymssqlconn() as conn:
            cursor = conn.cursor()
            cursor.execute("SELECT @@version AS version")
            self.assertIsNotNone(conn._conn)

        with self.assertRaises(InterfaceError) as context:
            self.assertIsNotNone(conn._conn)

        self.assertEqual(str(context.exception), "Connection is closed.")

def test_pymssql_Cursor_with(self):
        conn = pymssqlconn()
        with conn.cursor() as cursor:
            cursor.execute("SELECT @@version AS version")
            self.assertIsNotNone(conn._conn)

        self.assertIsNotNone(cursor)

        with self.assertRaises(InterfaceError) as context:
            cursor.execute("SELECT @@version AS version")

        self.assertEqual(str(context.exception), "Cursor is closed.")

def test_gevent_socket_mssql_execute_wait_read_concurrency(self):
        def wait_callback(read_fileno):
            gevent.socket.wait_read(read_fileno)

        pymssql.set_wait_callback(wait_callback)

        elapsed_time = self._run_all_greenlets(
            self.greenlet_run_mssql_execute)

        self.assertTrue(
            elapsed_time < datetime.timedelta(seconds=20),
            'elapsed_time < 20 seconds')

def test_gevent_socket_pymssql_execute_wait_read_concurrency(self):
        def wait_callback(read_fileno):
            gevent.socket.wait_read(read_fileno)

        pymssql.set_wait_callback(wait_callback)

        elapsed_time = self._run_all_greenlets(
            self.greenlet_run_pymssql_execute)

        self.assertTrue(
            elapsed_time < datetime.timedelta(seconds=20),
            'elapsed_time < 20 seconds')

def testBigIntPymssql(self):
        """Same as testBigInt above but from pymssql. Uses pymssql.output class."""

        if sys.version_info >= (3, ):
            py_type = int
        else:
            py_type = long

        in_val = 123456789
        cursor = self.pymssql.cursor()
        retval = cursor.callproc('pymssqlTestBigInt', [in_val, pymssql.output(py_type)])
        eq_(in_val, retval[1])

        in_val = 2147483647
        retval = cursor.callproc('pymssqlTestBigInt', [in_val, pymssql.output(py_type)])
        eq_(in_val, retval[1])