-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdb_insert
More file actions
138 lines (112 loc) · 4.23 KB
/
db_insert
File metadata and controls
138 lines (112 loc) · 4.23 KB
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
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
import sqlite3
def addDeveloper(name, joiningDate,id=False):
try:
sqliteConnection = sqlite3.connect('SQLite_Python.db',
detect_types=sqlite3.PARSE_DECLTYPES |
sqlite3.PARSE_COLNAMES)
cursor = sqliteConnection.cursor()
print("Connected to SQLite")
# sqlite_create_table_query = '''CREATE TABLE new_developers (
# id INTEGER PRIMARY KEY NOT NULL,
# name TEXT NOT NULL,
# joiningDate timestamp);'''
#
# cursor = sqliteConnection.cursor()
# cursor.execute(sqlite_create_table_query)
if id:
# insert developer detail
data_tuple = (id, name, joiningDate)
sqlite_insert_with_param = """INSERT INTO 'new_developers'
('id', 'name', 'joiningDate')
VALUES (?, ?, ?);"""
cursor.execute(sqlite_insert_with_param, data_tuple)
sqliteConnection.commit()
print("Developer added successfully \n")
else:
sqlite_insert_no_id = """INSERT INTO 'new_developers'
('name', 'joiningDate')
VALUES (?, ?);"""
data_tuple = (name, joiningDate)
cursor.execute(sqlite_insert_no_id, data_tuple)
sqliteConnection.commit()
print("Developer added successfully \n")
# get developer detail
sqlite_select_query = """SELECT name, joiningDate from new_developers where id = ?,? """
cursor.execute(sqlite_select_query, (1,2))
records = cursor.fetchall()
for row in records:
developer= row[0]
joining_Date = row[1]
print(developer, " joined on", joining_Date)
print("joining date type is", type(joining_Date))
cursor.close()
except sqlite3.Error as error:
print("Error while working with SQLite > ", error)
finally:
if (sqliteConnection):
sqliteConnection.close()
print("sqlite connection is closed")
try:
key = 1
addDeveloper(name='John', joiningDate=datetime.datetime.now())
except:
sqlite3.InternalError
from datetime import datetime
today = datetime.now().isoformat(sep=" ")[:19]
def createtable(tablename):
db = sqlite3.connect("static/app.db")
cursor = db.cursor()
try:
cursor.execute('''create TABLE activities(name text, date_from text,date_to text, location text, people text, notes text ,created text)''')
db.commit()
print("Table created")
except Exception as e:
db.rollback()
print(e)
print("Failed")
db.close()
return
def createtable2(tablename):
db = sqlite3.connect("static/app.db")
cursor = db.cursor()
try:
cursor.execute('''CREATE TABLE action_log(action_id integer primary key not null,action text, date_from text,date_to text, location text, people text, notes text ,created text)''')
db.commit()
print("Table created")
except Exception as e:
db.rollback()
print(e)
print("Failed")
db.close()
return
def writetodb(activity):
db = sqlite3.connect("static/app.db")
cursor = db.cursor()
try:
cursor.execute("INSERT INTO activities VALUES(?,?,?,?,?,?,?)",activity)
db.commit()
print("Commit succes")
except Exception as e:
db.rollback()
print(e)
print("Commit failed")
db.close()
return
def delfromdb(table="activities",column='',query=''):
db = sqlite3.connect("static/app.db")
cursor = db.cursor()
try:
cursor = cursor.execute(f'''DELETE FROM {table} WHERE "{column}" LIKE "%{query}%" ''')
db.commit()
print("DB Succes")
return
except Exception as e:
db.rollback()
print(e)
print("DB update failed")
db.close()
def readfromdb(table=None , column='' , query=''):
db = sqlite3.connect("static/app.db")
cursor = db.cursor()
cursor.execute(f"SELECT DISTINCT * FROM {table} WHERE {column} LIKE '%{query}%' ")
return [x for x in cursor]