Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_connection_execute(self, con):
result = con.execute("drop table if exists FOO;")
result = con.execute("create table FOO (a int);")
assert isinstance(result, Cursor)
con.execute("drop table if exists FOO;")
def test_empty_iterable(self):
c = Cursor(None)
result = list(c)
assert result == []
def test_arraysize(self):
c = Cursor(None)
assert c.arraysize == 1
c.arraysize = 10
assert c.arraysize == 10
with pytest.raises(TypeError):
c.arraysize = 'a'
def execute(self, operation, parameters=None):
"""Execute a SQL statement
Parameters
----------
operation: str
A SQL statement to exucute
Returns
-------
c: Cursor
"""
c = Cursor(self)
return c.execute(operation, parameters=parameters)
def cursor(self):
"""Create a new :class:`Cursor` object attached to this connection."""
return Cursor(self)
def to_df(self):
"""Convert the cursor to a data frame.
Returns
-------
dataframe : pandas.DataFrame
"""
if isinstance(self.cursor, Cursor):
col_names = [c.name for c in self.cursor.description]
result = pd.DataFrame(self.cursor.fetchall(), columns=col_names)
elif self.cursor is None:
result = pd.DataFrame([])
else:
result = self.cursor
return result