io
In this section, you will find information on how to add new types of *Connection
objects, *File
objects, or *Manager
objects.
Connection
Each *Connection
object inherits from the abstract Connection
class, which forces each type of connection object to
accept these arguments when creating the object:
args
, various positional argumentskwargs
, various keyword arguments
Besides this, the class must have a connect
and a close
method, respectively to connect to the database and one to close the connection,
respectively.
class Connection(ABC):
"""Connection base class"""
def __init__(self, *args, **kwargs):
"""Connection base object."""
self.connection = None
self.cursor = None
self.args = args
self.kwargs = kwargs
@abstractmethod
def connect(self):
pass
@abstractmethod
def close(self):
pass
def __bool__(self):
return True if self.connection and self.cursor else False
def __repr__(self):
return f"<{self.__class__.__name__} object, connection={self.connection}, cursor={self.cursor}>"
def __iter__(self):
if self.cursor:
return (e for e in self.cursor)
else:
return iter([])
Example Connection
based class:
class SQLiteConnection(Connection):
"""Connection sqlite class"""
def connect(self):
self.connection = sqlite3.connect(*self.args, **self.kwargs)
self.cursor = self.connection.cursor()
def close(self):
self.connection.close()
self.cursor.close()
Warning
All connections are DBAPI 2.0 compliant. If you need to create your own, it must adhere to these APIs.
File
The File
is the abstract class that the other *File
classes are based on.
It contains only the file
attribute, where the path of the file is saved during the creation of the object and two methods:
read
to read the contents of the file (must return a Dataset object) and write
(accept a Dataset) and writes to the destination file.
class File(ABC):
"""File base class"""
def __init__(self, filename):
"""File base object
:param filename: file path
"""
self.file = filename
@abstractmethod
def write(self, data):
"""Write data on file
:param data: data to write on file
:return: None
"""
pass
@abstractmethod
def read(self, **kwargs):
"""Read with format
:return: Dataset object
"""
pass
def __bool__(self):
return True if self.file else False
def __repr__(self):
return f"<{self.__class__.__name__} object, file={self.file}>"
def __iter__(self):
with open(self.file) as file:
for line in file:
yield line
Example File
based class:
class CsvFile(File):
"""CSV file class"""
def write(self, data):
"""Write data on csv file
:param data: data to write on csv file
:return: None
"""
if not isinstance(data, tablib.Dataset):
data = tablib.Dataset(data)
with open(self.file, mode="w") as file:
file.write(data.export("csv"))
def read(self, **kwargs):
"""Read csv format
:return: Dataset object
"""
with open(self.file) as file:
return tablib.Dataset().load(file, **kwargs)
Alias
When creating a Connection
or File
class, if you want to use the manager
function to create the returning *Manager
object,
you need to create an alias. There are two dicts in the io
module, which represent the aliases of these objects.
If you have created a new Connection
class, you will need to enter your alias in the DBTYPE
dict while for File-type classes,
enter it in the FILETYPE
dict. Here is an example: 'ods': ODSFile
Manager
Managers are classes that represent an input and output manager. For example, the DatabaseManager
class accepts a
Connection
object and implements methods on these types of objects representing database connections.
class DatabaseManager(Manager):
"""Database manager class for SQL connection"""
def __init__(self, connection: Connection):
"""Database manager object for SQL connection
:param connection: Connection based object
"""
self.type = "sql"
self.connector = connection
# Connect database
self.connector.connect()
# Set description
self.description = None
self.data = None
# Row properties
self.lastrowid = None
self.rowcount = 0
def __repr__(self):
"""Representation of DatabaseManager object
:return: string
"""
ret = f"<{self.__class__.__name__} object, "
ret += f"connection={self.connector.__class__.__name__}>"
return ret
def __iter__(self):
if self.connector.cursor:
return (e for e in self.connector.cursor)
else:
return iter([])
def reconnect(self):
"""Close and start connection
:return: None
"""
# Close connection
self.connector.close()
# Start connection, again
self.connector.connect()
def execute(self, query, params=None):
"""Execute query on database cursor
:param query: SQL query language
:param params: parameters of the query
:return: None
"""
if params:
self.connector.cursor.execute(query, params)
else:
self.connector.cursor.execute(query)
# Set last row id
self.lastrowid = self.connector.cursor.lastrowid
# Set row cont
self.rowcount = self.connector.cursor.rowcount
# Set description
self.description = self.connector.cursor.description
def executemany(self, query, params):
"""Execute query on database cursor with many parameters
:param query: SQL query language
:param params: list of parameters of the query
:return: None
"""
# See if query was cached
self.connector.cursor.executemany(query, params)
# Set last row id
self.lastrowid = self.connector.cursor.lastrowid
# Set row cont
self.rowcount = self.connector.cursor.rowcount
# Set description
self.description = self.connector.cursor.description
def fetchall(self) -> tablib.Dataset:
"""Fetches all (or all remaining) rows of a query result set
:return: Dataset object
"""
header = [field[0] for field in self.description]
self.data = tablib.Dataset(headers=header)
for row in self.connector.cursor.fetchall():
self.data.append(list(row))
return self.data
def fetchone(self) -> tablib.Dataset:
"""Retrieves the next row of a query result set
:return: Dataset object
"""
header = [field[0] for field in self.description]
self.data = tablib.Dataset(
list(self.connector.cursor.fetchone()), headers=header
)
return self.data
def fetchmany(self, size=1) -> tablib.Dataset:
"""Fetches the next set of rows of a query result
:param size: the number of rows returned
:return: Dataset object
"""
header = [field[0] for field in self.description]
self.data = tablib.Dataset(headers=header)
for row in self.connector.cursor.fetchmany(size):
self.data.append(list(row))
return self.data
def callproc(self, proc_name, params=None) -> tablib.Dataset:
"""Calls the stored procedure named
:param proc_name: name of store procedure
:param params: sequence of parameters must contain one entry for each argument that the procedure expects
:return: Dataset object
"""
if params is None:
params = []
header = [field[0] for field in self.description]
self.data = tablib.Dataset(headers=header)
for row in self.connector.cursor.callproc(proc_name, params):
self.data.append(list(row))
return self.data
def commit(self):
"""This method sends a COMMIT statement to the server
:return: None
"""
self.connector.connection.commit()
Manager function
Each *Manager
class has associated a function of type create_<type of manager>_manager(*args, **kwargs)
.
This function will then be used by the manager
function to create the corresponding *Manager
object based on its alias.
For example, the DatabaseManager
class has associated the create_database_manager
function which will be called by the
manager
function to create the object based on the type of alias passed.
def manager(datatype, *args, **kwargs):
"""Creates manager object based on datatype
:param datatype: type of manager
:param args: various positional arguments
:param kwargs: various keyword arguments
:return: Manager object
"""
# Choose manager type
if datatype in DBTYPE:
return create_database_manager(datatype, *args, **kwargs)
elif datatype in FILETYPE:
return create_file_manager(datatype, *args, **kwargs)
elif datatype == "ldap":
return create_ldap_manager(*args, **kwargs)
elif datatype == "nosql":
connection = kwargs.get("connection") or args[0]
nargs = args[1:]
try:
kwargs.pop("connection")
except KeyError:
pass
return create_nosql_manager(connection, *nargs, **kwargs)
else:
raise ValueError(f"data type {datatype} doesn't exists!")
def create_database_manager(dbtype, *args, **kwargs):
"""Creates a DatabaseManager object
:param dbtype: type of database connection
:return: DatabaseManager
"""
# Create DatabaseManager object
connection = DBTYPE[dbtype](*args, **kwargs)
return DatabaseManager(connection=connection)
Example
Here we will see how to create your own *Connection
class to access a specific database.
import pyreports
import DB2
# class for connect DB2 database
class DB2Connection(pyreports.io.Connection):
def connect(self):
self.connection = DB2.connect(*self.args, **self.kwargs)
self.cursor = self.connection
def close(self):
self.connection.close()
self.cursor.close()
# Create an alias for DB2Connection object
pyreports.io.DBTYPE['db2'] = DB2Connection
# Create my DatabaseManager object
mydb2 = pyreports.manager('db2', dsn='sample', uid='db2inst1', pwd='ibmdb2')