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 arguments

• kwargs, 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')