-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdb_app_funtions.py
70 lines (48 loc) · 1.79 KB
/
db_app_funtions.py
1
2
3
4
5
6
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
import numpy as np
import pandas as pd
import sqlite3
import io
import base64
def parse_content(contents, filename):
print("Received contents:", contents)
content_type, content_string = contents.split(',')
decoded = base64.b64decode(content_string)
if '.csv' in filename:
df = pd.read_csv(io.StringIO(decoded.decode('utf-8')))
return df.to_dict('records')
elif '.xls' in filename:
df = pd.read_excel(io.BytesIO(decoded))
return df.to_dict('records')
def decode_image(image_file):
encoded = base64.b64encode(open(image_file, 'rb').read())
return f"data:image/png;base64,{encoded.decode()}"
def connect_pd_database():
db_path = 'C:\sqlite\Databases\ifrs9test.db'
pd_table = 'ifrs9_PD_data'
recoveries_table = 'ifrs9_recoveries_data'
# loan_table = 'ifrs9_loanbook'
connection = sqlite3.connect(db_path)
query1 = f'SELECT * FROM {pd_table}'
query2 = f'SELECT * FROM {recoveries_table}'
# query3 = f'SELECT * FROM {loan_table}'
pd_data = pd.read_sql_query(query1, connection)
rec_data = pd.read_sql_query(query2, connection)
# loan_book =pd.read_sql_query(query3, connection)
connection.close()
return pd_data, rec_data#, loan_book
def connect_loanbook_database():
db_path = 'C:\sqlite\Databases\ifrs9loanbook.db'
loan_book_table = 'ifrs9_loanbook'
connection = sqlite3.connect(db_path)
query1 = f'SELECT * FROM {loan_book_table}'
loan_book = pd.read_sql_query(query1, connection)
connection.close()
return loan_book
def connect_fli():
db_path = 'C:\sqlite\Databases\ifrs9fli.db'
fli_table = 'fli'
connection = sqlite3.connect(db_path)
query1 = f'SELECT * FROM {fli_table}'
fli = pd.read_sql_query(query1, connection)
connection.close()
return fli