Projekt

Obecné

Profil

Stáhnout (4.15 KB) Statistiky
| Větev: | Tag: | Revize:
1
from sqlite3 import Connection, Cursor, Error
2
from typing import List
3

    
4
from injector import inject
5

    
6
from src.exceptions.database_exception import DatabaseException
7
from src.model.private_key import PrivateKey
8
from src.constants import *
9

    
10

    
11
class PrivateKeyRepository:
12

    
13
    @inject
14
    def __init__(self, connection: Connection):
15
        """
16
        Constructor of the PrivateKeyRepository object
17

    
18
        :param connection: Instance of the Connection object
19
        :param cursor: Instance of the Cursor object
20
        """
21

    
22
        self.connection = connection
23
        self.cursor = connection.cursor()
24

    
25
    def create(self, private_key: PrivateKey):
26
        """
27
        Creates a private key.
28

    
29
        :param private_key: Instance of the PrivateKey object
30

    
31
        :return: the result of whether the creation was successful
32
        """
33

    
34
        try:
35
            sql = (f"INSERT INTO {TAB_PRIVATE_KEYS} "
36
                   f"({COL_PRIVATE_KEY},"
37
                   f"{COL_PASSWORD}) "
38
                   f"VALUES(?,?)")
39
            values = [private_key.private_key,
40
                      private_key.password]
41
            self.cursor.execute(sql, values)
42
            last_id = self.cursor.lastrowid
43
            self.connection.commit()
44
        except Error as e:
45
            raise DatabaseException(e)
46

    
47
        return last_id
48

    
49
    def read(self, private_key_id: int):
50
        """
51
        Reads (selects) a private key.
52

    
53
        :param private_key_id: ID of specific private key
54

    
55
        :return: instance of the PrivateKey object
56
        """
57

    
58
        try:
59
            sql = (f"SELECT * FROM {TAB_PRIVATE_KEYS} "
60
                   f"WHERE {COL_ID} = ?")
61
            values = [private_key_id]
62
            self.cursor.execute(sql, values)
63
            private_key_row = self.cursor.fetchone()
64

    
65
            if private_key_row is None:
66
                return None
67

    
68
            private_key: PrivateKey = PrivateKey(private_key_row[0],
69
                                                 private_key_row[1],
70
                                                 private_key_row[2])
71
        except Error as e:
72
            raise DatabaseException(e)
73

    
74
        return private_key
75

    
76
    def read_all(self):
77
        """
78
        Reads (selects) all private keys.
79

    
80
        :return: list of private keys
81
        """
82

    
83
        try:
84
            sql = f"SELECT * FROM {TAB_PRIVATE_KEYS}"
85
            self.cursor.execute(sql)
86
            private_key_rows = self.cursor.fetchall()
87

    
88
            private_keys: List[PrivateKey] = []
89
            for private_key_row in private_key_rows:
90
                private_keys.append(PrivateKey(private_key_row[0],
91
                                               private_key_row[1],
92
                                               private_key_row[2]))
93
        except Error as e:
94
            raise DatabaseException(e)
95

    
96
        return private_keys
97

    
98
    def update(self, private_key_id: int, private_key: PrivateKey) -> bool:
99
        """
100
        Updates a private key.
101

    
102
        :param private_key_id: ID of specific private key
103
        :param private_key: Instance of the PrivateKey object
104

    
105
        :return: the result of whether the updation was successful
106
        """
107

    
108
        try:
109
            sql = (f"UPDATE {TAB_PRIVATE_KEYS} "
110
                   f"SET {COL_PRIVATE_KEY} = ?, "
111
                   f"{COL_PASSWORD} = ? "
112
                   f"WHERE {COL_ID} = ?")
113
            values = [private_key.private_key,
114
                      private_key.password,
115
                      private_key_id]
116
            self.cursor.execute(sql, values)
117
            self.connection.commit()
118
        except Error as e:
119
            raise DatabaseException(e)
120

    
121
        return self.cursor.rowcount > 0
122

    
123
    def delete(self, private_key_id: int) -> bool:
124
        """
125
        Deletes a private key
126

    
127
        :param private_key_id: ID of specific private key
128

    
129
        :return: the result of whether the deletion was successful
130
        """
131

    
132
        try:
133
            sql = (f"DELETE FROM {TAB_PRIVATE_KEYS} "
134
                   f"WHERE {COL_ID} = ?")
135
            values = [private_key_id]
136
            self.cursor.execute(sql, values)
137
            self.connection.commit()
138
        except Error as e:
139
            raise DatabaseException
140

    
141
        return self.cursor.rowcount > 0
(3-3/3)