Skip to content

Common

Common modules

For example testing database connection and CRUD we use sqlite3 become_qa_auto.db

This module defines the class Database to connect & manipulate DB data.

Database

Defines several methods to connect & CRUD DB data.

Source code in modules/common/database.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class Database:
    """Defines several methods to connect & CRUD DB data."""

    def __init__(self) -> None:
        """Set connection to DB file."""
        self.connection = sqlite3.connect(config_dict["DB"]["path"])
        self.cursor = self.connection.cursor()

    def test_connection(self) -> None:
        """Test connection to DB.

        Returns:
            print sqlite version
        """
        sqlite_select_Query = "SELECT sqlite_version();"
        self.cursor.execute(sqlite_select_Query)
        record = self.cursor.fetchall()
        print(f"Connected successfully. SQLite Database Version is: {record}")

    def get_all_users(self) -> list:
        """Get the rows of a query result all customers from DB.

        Returns:
            List of tuples with data
        """
        query = "SELECT name, address, city FROM customers"
        self.cursor.execute(query)
        record = self.cursor.fetchall()
        return record

    def get_user_address_by_name(self, name: str) -> list:
        """Get data by name from DB.

        Args:
            name: Customer's name

        Returns:
            List of tuples with data
        """
        query = f"SELECT address, city, postalCode, country FROM customers \
            WHERE name = '{name}'"
        self.cursor.execute(query)
        record = self.cursor.fetchall()
        return record

    def update_product_qnt_by_id(self, product_id: int, qnt: int) -> None:
        """Update quantity product by ID.

        Args:
            product_id: product ID
            qnt: product quantity
        """
        query = f"UPDATE products SET quantity = {qnt} WHERE id = {product_id}"
        self.cursor.execute(query)
        self.connection.commit()

    def select_product_qnt_by_id(self, product_id: int) -> list:
        """Get quantity product by ID

        Args:
            product_id: product ID

        Returns:
            List of tuples with data
        """
        query = f"SELECT quantity FROM products WHERE id = {product_id}"
        self.cursor.execute(query)
        record = self.cursor.fetchall()
        return record

    def insert_product(
        self, product_id: int, name: str, description: str, qnt: int
    ) -> None:
        """Inserts or replace product

        Args:
            product_id: product ID
            name: name of the product
            description: Some text for the description
            qnt: quantity of the product
        """
        query = f"INSERT OR REPLACE INTO products \
            (id, name, description, quantity) \
            VALUES ({product_id}, '{name}', '{description}', {qnt})"
        self.cursor.execute(query)
        self.connection.commit()

    def delete_product_by_id(self, product_id: int) -> None:
        """Deleting product by ID.

        Args:
            product_id: product ID
        """
        query = f"DELETE FROM products WHERE id = {product_id}"
        self.cursor.execute(query)
        self.connection.commit()

    def get_detailed_orders(self) -> list:
        """Get detailed info about orders with JOIN.

        Returns:
            List of tuples with data
        """
        query = "SELECT orders.id, customers.name, products.name, \
            products.description, orders.order_date \
            FROM orders \
            JOIN customers ON orders.customer_id = customers.id \
            JOIN products ON orders.product_id = products.id"
        self.cursor.execute(query)
        record = self.cursor.fetchall()
        return record

__init__()

Set connection to DB file.

Source code in modules/common/database.py 
10
11
12
13
def __init__(self) -> None:
    """Set connection to DB file."""
    self.connection = sqlite3.connect(config_dict["DB"]["path"])
    self.cursor = self.connection.cursor()

delete_product_by_id(product_id)

Deleting product by ID.

Parameters:

Name Type Description Default
product_id int

product ID

 required
Source code in modules/common/database.py 
 94
 95
 96
 97
 98
 99
100
101
102
def delete_product_by_id(self, product_id: int) -> None:
    """Deleting product by ID.

    Args:
        product_id: product ID
    """
    query = f"DELETE FROM products WHERE id = {product_id}"
    self.cursor.execute(query)
    self.connection.commit()

get_all_users()

Get the rows of a query result all customers from DB.

Returns:

Type Description
list

List of tuples with data
Source code in modules/common/database.py 
26
27
28
29
30
31
32
33
34
35
def get_all_users(self) -> list:
    """Get the rows of a query result all customers from DB.

    Returns:
        List of tuples with data
    """
    query = "SELECT name, address, city FROM customers"
    self.cursor.execute(query)
    record = self.cursor.fetchall()
    return record

get_detailed_orders()

Get detailed info about orders with JOIN.

Returns:

Type Description
list

List of tuples with data
Source code in modules/common/database.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def get_detailed_orders(self) -> list:
    """Get detailed info about orders with JOIN.

    Returns:
        List of tuples with data
    """
    query = "SELECT orders.id, customers.name, products.name, \
        products.description, orders.order_date \
        FROM orders \
        JOIN customers ON orders.customer_id = customers.id \
        JOIN products ON orders.product_id = products.id"
    self.cursor.execute(query)
    record = self.cursor.fetchall()
    return record

get_user_address_by_name(name)

Get data by name from DB.

Parameters:

Name Type Description Default
name str

Customer's name

 required

Returns:

Type Description
list

List of tuples with data
Source code in modules/common/database.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def get_user_address_by_name(self, name: str) -> list:
    """Get data by name from DB.

    Args:
        name: Customer's name

    Returns:
        List of tuples with data
    """
    query = f"SELECT address, city, postalCode, country FROM customers \
        WHERE name = '{name}'"
    self.cursor.execute(query)
    record = self.cursor.fetchall()
    return record

insert_product(product_id, name, description, qnt)

Inserts or replace product

Parameters:

Name Type Description Default
product_id int

product ID

 required
name str

name of the product

 required
description str

Some text for the description

 required
qnt int

quantity of the product

 required
Source code in modules/common/database.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def insert_product(
    self, product_id: int, name: str, description: str, qnt: int
) -> None:
    """Inserts or replace product

    Args:
        product_id: product ID
        name: name of the product
        description: Some text for the description
        qnt: quantity of the product
    """
    query = f"INSERT OR REPLACE INTO products \
        (id, name, description, quantity) \
        VALUES ({product_id}, '{name}', '{description}', {qnt})"
    self.cursor.execute(query)
    self.connection.commit()

select_product_qnt_by_id(product_id)

Get quantity product by ID

Parameters:

Name Type Description Default
product_id int

product ID

 required

Returns:

Type Description
list

List of tuples with data
Source code in modules/common/database.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
def select_product_qnt_by_id(self, product_id: int) -> list:
    """Get quantity product by ID

    Args:
        product_id: product ID

    Returns:
        List of tuples with data
    """
    query = f"SELECT quantity FROM products WHERE id = {product_id}"
    self.cursor.execute(query)
    record = self.cursor.fetchall()
    return record

test_connection()

Test connection to DB.

Returns:

Type Description
None

print sqlite version
Source code in modules/common/database.py 
15
16
17
18
19
20
21
22
23
24
def test_connection(self) -> None:
    """Test connection to DB.

    Returns:
        print sqlite version
    """
    sqlite_select_Query = "SELECT sqlite_version();"
    self.cursor.execute(sqlite_select_Query)
    record = self.cursor.fetchall()
    print(f"Connected successfully. SQLite Database Version is: {record}")

update_product_qnt_by_id(product_id, qnt)

Update quantity product by ID.

Parameters:

Name Type Description Default
product_id int

product ID

 required
qnt int

product quantity

 required
Source code in modules/common/database.py 
52
53
54
55
56
57
58
59
60
61
def update_product_qnt_by_id(self, product_id: int, qnt: int) -> None:
    """Update quantity product by ID.

    Args:
        product_id: product ID
        qnt: product quantity
    """
    query = f"UPDATE products SET quantity = {qnt} WHERE id = {product_id}"
    self.cursor.execute(query)
    self.connection.commit()