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 SQLliteConnection(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 pyreports.io.DatabaseManager(connection: Connection)
Database manager class for SQL connection
- callproc(proc_name, params=None) Dataset
Calls the stored procedure named
- Parameters
proc_name – name of store procedure
params – sequence of parameters must contain one entry for each argument that the procedure expects
- Returns
Dataset object
- commit()
This method sends a COMMIT statement to the server
- Returns
None
- execute(query, params=None)
Execute query on database cursor
- Parameters
query – SQL query language
params – parameters of the query
- Returns
None
- executemany(query, params)
Execute query on database cursor with many parameters
- Parameters
query – SQL query language
params – list of parameters of the query
- Returns
None
- fetchall() Dataset
Fetches all (or all remaining) rows of a query result set
- Returns
Dataset object
- fetchmany(size=1) Dataset
Fetches the next set of rows of a query result
- Parameters
size – the number of rows returned
- Returns
Dataset object
- fetchone() Dataset
Retrieves the next row of a query result set
- Returns
Dataset object
- reconnect()
Close and start connection
- Returns
None
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.
- pyreports.io.manager(datatype, *args, **kwargs)
Creates manager object based on datatype
- Parameters
datatype – type of manager
args – various positional arguments
kwargs – various keyword arguments
- Returns
Manager object
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')