bbarchivist.sqlutils

Complexity

Total Complexity

16

Size/Duplication

Total Lines	151
Duplicated Lines	0 %

Test Coverage

Coverage

100%

Importance

Changes

0

7 Functions

Rating	Name	Duplication	Size	Complexity
A	list_sw_releases()	0	18	3
A	export_sql_db()	0	17	2
A	insert()	0	32	4
A	pop_sw_release()	0	15	2
A	prepare_sw_db()	0	14	2
A	check_exists()	0	19	2
A	prepare_path()	0	6	1

#!/usr/bin/env python3
"""This module is used for dealing with SQL databases, including CSV export."""

import csv  # write to csv
import operator  # for sorting
import os  # paths
import sqlite3  # the sql library
import time  # current date

from bbarchivist import decorators  # sql handlers
from bbarchivist import iniconfig  # config directory

__author__ = "Thurask"
__license__ = "WTFPL v2"
__copyright__ = "2015-2019 Thurask"


def prepare_path():
    """
    Figure out where the path is.
    """
    sqlpath = os.path.join(iniconfig.config_homepath(None), "bbarchivist.db")
    return sqlpath


@decorators.sql_excepthandler("False")
def prepare_sw_db():
    """
    Create SQLite database, if not already existing.
    """
    cnxn = sqlite3.connect(prepare_path())
    with cnxn:
        crs = cnxn.cursor()
        reqid = "INTEGER PRIMARY KEY"
        reqs = "TEXT NOT NULL UNIQUE COLLATE NOCASE"
        reqs2 = "TEXT NOT NULL"
        table = "Swrelease(Id {0}, Os {1}, Software {1}, Available {2}, Date {2})".format(
            reqid, reqs, reqs2)
        crs.execute("CREATE TABLE IF NOT EXISTS " + table)


@decorators.sql_excepthandler("True")
def insert(osversion, swrelease, available, curdate=None):
    """
    Insert values into main SQLite database.

    :param osversion: OS version.
    :type osversion: str

    :param swrelease: Software release.
    :type swrelease: str

    :param servers: If release is available. String converted boolean.
    :type servers: str

    :param curdate: If None, today. For manual dates, specify this.
    :type curdate: str
    """
    if curdate is None:
        curdate = time.strftime("%Y %B %d")
    cnxn = sqlite3.connect(prepare_path())
    with cnxn:
        crs = cnxn.cursor()
        try:  # insert if new
            crs.execute(
                "INSERT INTO Swrelease(Os, Software, Available, Date) VALUES (?,?,?,?)",
                (osversion,
                 swrelease,
                 available,
                 curdate))
        except sqlite3.IntegrityError:  # update if not new
            crs.execute("UPDATE Swrelease SET Available=? WHERE Os=? AND Software=?",
                        (available, osversion, swrelease))


@decorators.sql_excepthandler("False")
def pop_sw_release(osversion, swrelease):
    """
    Remove given entry from database.

    :param osversion: OS version.
    :type osversion: str

    :param swrelease: Software release.
    :type swrelease: str
    """
    cnxn = sqlite3.connect(prepare_path())
    with cnxn:
        crs = cnxn.cursor()
        crs.execute("DELETE FROM Swrelease WHERE Os=? AND Software=?", (osversion, swrelease))


@decorators.sql_excepthandler("False")
def check_exists(osversion, swrelease):
    """
    Check if we did this one already.

    :param osversion: OS version.
    :type osversion: str

    :param swrelease: Software release.
    :type swrelease: str
    """
    cnxn = sqlite3.connect(prepare_path())
    with cnxn:  # check if exists
        crs = cnxn.cursor()
        exis = crs.execute(
            "SELECT EXISTS (SELECT 1 FROM Swrelease WHERE Os=? AND Software=?)",
            (osversion,
             swrelease)).fetchone()[0]
        return bool(exis)


@decorators.sql_excepthandler("False")
@decorators.sql_existhandler(prepare_path())
def export_sql_db():
    """
    Export main SQL database into a CSV file.
    """
    cnxn = sqlite3.connect(prepare_path())
    with cnxn:
        csvpath = os.path.join(os.path.expanduser("~"), "swrelease.csv")
        csvw = csv.writer(open(csvpath, "w"), dialect='excel')
        crs = cnxn.cursor()
        crs.execute("SELECT Os, Software, Available, Date FROM Swrelease")
        rows = crs.fetchall()
        sortedrows = sorted(rows, key=operator.itemgetter(0))
        csvw.writerow(('OS Version', 'Software Release', 'Available', 'Date Modified'))
        csvw.writerows(sortedrows)
    print("EXPORTED!\n{0}".format(csvpath))


@decorators.sql_excepthandler("False")
@decorators.sql_existhandler(prepare_path())
def list_sw_releases(avail=False):
    """
    Return every SW/OS pair in the database.

    :param avail: If we filter out non-available results. Default is false.
    :type avail: bool
    """
    cnxn = sqlite3.connect(prepare_path())
    with cnxn:
        crs = cnxn.cursor()
        query = "SELECT Os, Software, Available, Date FROM Swrelease"
        if avail:
            query += " WHERE Available= 'available'"
        crs.execute(query)
        rows = crs.fetchall()
        return rows