requests.py 7.28 KB
Newer Older
Leo Peron's avatar
Leo Peron committed
1
2


Leo Peron's avatar
Leo Peron committed
3
4
import utils
import psycopg2 as sql
Leo Peron's avatar
Leo Peron committed
5
6
7
from psycopg2._psycopg import AsIs
from datetime import datetime as dt
from typing import List
Leo Peron's avatar
Leo Peron committed
8
9
10
11
12
13
14


class Requests:
    def __init__(self):
        self.utils = utils.Utils()
        self.datas = self.utils.loadDatas()
        self.conn = self.__connect()
Leo Peron's avatar
Leo Peron committed
15
        self.conn.autocommit = True
Leo Peron's avatar
Leo Peron committed
16
17
18
        self.cur = self.__getCursor()
        self.account_state = ["Ouvert", "Bloqué", "Fermé"]
        self.operation_state = ["Traitée", "Non Traitée"]
Leo Peron's avatar
Leo Peron committed
19
        self.account_type = ["courant", "revolving", "epargne"]
Leo Peron's avatar
Leo Peron committed
20
21
22
23
24
25
26
27
28
29
30
31

    def __connect(self):
        try:
            conn = sql.connect(
                f'host={self.datas["host"]} dbname={self.datas["dbname"]} user={self.datas["user"]} password={self.datas["pswd"]}'
            )
            return conn
        except sql.Error as e:
            print("Erreur de connexion")
            return None

    def __getCursor(self):
Leo Peron's avatar
Leo Peron committed
32
        if self.conn:
Leo Peron's avatar
Leo Peron committed
33
34
35
36
            return self.conn.cursor()
        else:
            return None

Leo Peron's avatar
Leo Peron committed
37
38
    def getUserByNum(self, num: int) -> List[str]:
        self.cur.execute("SELECT * FROM clients WHERE telephone = %s", (num,))
Leo Peron's avatar
Leo Peron committed
39
40
        return self.cur.fetchone()

Leo Peron's avatar
Leo Peron committed
41
    def getUserAccountsId(self, num: int):
Leo Peron's avatar
Leo Peron committed
42
        self.cur.execute(
Leo Peron's avatar
Leo Peron committed
43
            "SELECT Courant, Revolving, Epargne FROM Appartenance WHERE client = %s", (num,)
Leo Peron's avatar
Leo Peron committed
44
        )
Leo Peron's avatar
Leo Peron committed
45
        return self.__getAccounts(self.cur.fetchall())
Leo Peron's avatar
Leo Peron committed
46
47
48
49
50
51
52
53

    def __getAccounts(self, ids):
        accounts = []
        for elem, id in zip(self.account_type, ids):
            self.cur.execute(f"SELECT * FROM {elem} WHERE id = {id}")
            accounts.append(self.cur.fetchone())
        return accounts

Leo Peron's avatar
Leo Peron committed
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
    # GESTION D'UN UTILISATEUR

    def createUser(self, num: int, prenom: str, nom: str) -> bool:
        try:
            self.cur.execute("INSERT INTO clients VALUES (%s,%s,%s)", (num, prenom, nom))
            return True
        except sql.Error as e:
            return False

    def modifyUser(self, num: int, prenom: str, nom: str) -> bool:
        try:
            self.cur.execute("UPDATE clients SET prenom=%s, nom=%s WHRER telephone=%s", (prenom, nom, num))
            return True
        except sql.Error as e:
            return False

    def deleteUser(self, num: int) -> bool:
        try:
            self.cur.execute("DELETE FROM clients WHERE telephone=%s", (num,))
            return True
        except sql.Error as e:
            return False

    # CREATION DES COMPTES BANCAIRES

    def createCourantAccount(self, num: int, solde: int, decouvert: int) -> bool:
        id = self.__generateAccountId()
        try:
            self.cur.execute('INSERT INTO comptescourant VALUES (%s, current_timestamp , %s,%s,%s,%s,%s)',
                             (id, self.account_state[0], solde, solde, solde, decouvert))
            self.addUserToAccount(num, id, "courant")
            return True
        except sql.Error as e:
            return False

    def createRevolvingAccount(self, num: int, solde: int, taux: int, montant: int) -> bool:
        id = self.__generateAccountId()
        try:
            self.cur.execute('INSERT INTO comptesepargne VALUES (%s, current_timestamp, %s,%s,%s,%s)',
                         (id, self.account_state[0], taux, montant, solde))
            self.addUserToAccount(num, id, "revolving")
            return True
        except sql.Error as e:
            print(e)
            return False

    def createEpargneAccount(self, num: int, interet: int, plafond: int, solde: int) -> bool:
        id = self.__generateAccountId()
        try:
            self.cur.execute('INSERT INTO ComptesEpargne VALUES (%s, current_timestamp, %s,%s,%s,%s)',
                         (id, self.account_state[0], interet, plafond, solde))
            self.addUserToAccount(num, id, "epargne")
            return True
        except sql.Error as e:
            print(e)
            return False

    # MODIFICATION DES COMPTES BANCAIRES

    def modifyCourantAccount(self, id: int, decouvert: int) -> bool:
        try:
            self.cur.execute('UPDATE comptescourant SET decouvert=%s WHERE id=%s', (decouvert, id))
            return True
        except sql.Error as e:
            return False

    def modifyRevolvingAccount(self, id: int, taux: int) -> bool:
        try:
            self.cur.execute('UPDATE comptesrevolving SET taux=%s WHERE id=%s', (id, taux))
            return True
        except sql.Error as e:
            return False

    def modifyEpargneAccount(self, num: int, interet: int) -> bool:
        try:
            self.cur.execute('UPDATE comptesrevolving SET interet=%s WHERE id=%s', (id, interet))
            return True
        except sql.Error as e:
            return False

    # SUPPRESION D'UN COMPTE

    def deleteAccount(self, id: int):
        pass

    # Insertion d'une opération

    def createOperation(self, compte: int, op_type: str, acc_type: str, montant: int):
        id = self.__generateOperationId()
        try:
            self.cur.execute('INSERT INTO %s (id, %s, montant, etat, date) VALUES (%s, %s, %s, %s, current_timestamp)',
                             (AsIs('operations' + op_type),AsIs(acc_type), id, compte, montant, self.operation_state[0]))
            self.__updateSoldById(compte, type, montant)
            return True
        except sql.Error as e:
            print(e)
            return False
        pass

    def operationCheque(self):
        pass

    def operationGuichet(self):
        pass
        pass


    # GESTION DES RELATIONS COMPTES - CLIENTS

    def addUserToAccount(self, num: int, id: int, type: str) -> bool:
        try:
            self.cur.execute('INSERT INTO appartenance (client, %s) VALUES (%s, %s)', (AsIs(type), num, id))
            return True
        except sql.Error as e:
            return False

    def removeUserFromAccount(self, num: int, id: int, type: str) -> bool:
        try:
            self.cur.execute('DELETE FROM appartenance WHERE client=%s AND %s=%s', (num, AsIs(type), id))
            return True
        except sql.Error as e:
            return False

    # GENERATIONS CLES COMPTES ET OPERATIONS

    def __generateAccountId(self) -> int:
        self.cur.execute('SELECT MAX(id) FROM (SELECT id FROM ComptesEpargne UNION SELECT id FROM ComptesRevolving UNION SELECT id FROM ComptesCourant) AS comptes')
        cle = self.cur.fetchone()
        if not cle[0]:
            return 1
        return cle[0] + 1

    def __generateOperationId(self) -> int:
        self.cur.execute('SELECT MAX(id) FROM (SELECT id FROM operationscartebleue UNION SELECT id FROM operationscheque UNION SELECT id FROM operationsguichet UNION SELECT id FROM operationsvirement) AS operation')
        cle = self.cur.fetchone()
        if not cle[0]:
            return 1
        return cle[0] + 1

Leo Peron's avatar
Leo Peron committed
193
194
    def getAccountById(self, id):
        self.cur.execute(
Leo Peron's avatar
Leo Peron committed
195
            "SELECT * FROM comptecourant, compterevolving, comptecourant WHERE Courant.id = {id} OR Revolving.id = {id} OR Epargne.id = {id}"
Leo Peron's avatar
Leo Peron committed
196
197
198
        )


Leo Peron's avatar
Leo Peron committed
199
200
201
202
203
204
205
206
207
208
    def __updateSoldById(self, id: int, type: str, var: int) -> bool:
        try:
            self.cur.execute("SELECT solde FROM %s WHERE id =%s", (AsIs(type), id))
            raw = self.cur.fetchone()
            self.cur.execute(
                "UPDATE %s SET solde=%s WHERE id=%s", (AsIs('comptes'+type), raw[0] + var, id)
            )
            return True
        except sql.Error as e:
            return False