In python code: Using the dataframe produced in question#2 and your Database class from Lab1, Byte Stream your dataframe into your Database object. Also, use your Query Builder to build the Insert Query for your byte stream insertion. class GreenhouseGas(NamedTuple): Gas: str Pre_1750: float Recent: float Absolute_increase_since_1750: float Percentage_increase_since_1750: float def __repr__(self): return f"Gas: {self.Gas}, Pre-1750: {self.Pre_1750}, Recent: {self.Recent}, Absolute increase since 1750: {self.Absolute_increase_since_1750}, Percentage increase since 1750: {self.Percentage_increase_since_1750}" def __lt__(self, other): return self.Recent < other.Recent def __eq__(self, other): return self.Gas == other.Gas @classmethod def from_list(cls, data): gas, pre_1750, recent, absolute_increase, percentage_increase = data return cls(gas, pre_1750, recent, absolute_increase, percentage_increase) def search_gas(gas_list, gas_name): for gas in gas_list: if gas.Gas == gas_name: return gas return None gas_list = [ GreenhouseGas.from_list(["Gas1", 1.0, 2.0, 3.0, 4.0]), GreenhouseGas.from_list(["Gas2", 5.0, 6.0, 7.0, 8.0]), GreenhouseGas.from_list(["Gas3", 9.0, 10.0, 11.0, 12.0]), ] # sort the list by Recent gas_list.sort() # search for a specific gas result = search_gas(gas_list, "Gas2") print(result) class Database: def __init__(self, dbName): self.dbName = dbName def connect(self): global sqliteConnection try: sqliteConnection = sqlite3.connect(self.dbName) cursor = sqliteConnection.cursor() print("Database created and Successfully Connected to SQLite") select_Query = "select sqlite_version();" cursor.execute(select_Query) record = cursor.fetchall() except sqlite3.Error as error: print("Error while connecting to sqlite", error) def table(self, query): global sqliteConnection try: cursor = sqliteConnection.cursor() cursor.execute(query) sqliteConnection.commit() print("SQLite table created") except sqlite3.Error as error: print("Table exists: ", error) def insert(self, query, df): global sqliteConnection try: cursor = sqliteConnection.cursor() for row in df.itertuples(): insert_sql = query.format(row[0], row[1]) cursor.execute(insert_sql) sqliteConnection.commit() print("Inserted successfully into table") except sqlite3.Error as error: print("Failed to insert: ", error) def search(self, query, value): global sqliteConnection cursor = sqliteConnection.cursor() sel = query.format(value) cursor.execute(sel) result = cursor.fetchall() return result[0][0] def delete(self, query, id): global sqliteConnection try: cursor = sqliteConnection.cursor() delete_query = query + str(id) cursor.execute(delete_query) sqliteConnection.commit() print("Record deleted successfully ") except sqlite3.Error as error: print("Failed to delete record from sqlite table", error) def query_builder(self, name, qType, colandType, dataType=[]): col = list(colandType) types = list(colandType.values()) if qType == ("TABLE" or "table" or "Table"): query = f"CREATE TABLE IF NOT EXISTS {name} " for i in range(len(col)): if col[i] == col[-1]: query += f"{col[i]} {types[i]})" else: query += f"({col[i]} {types[i]}, " elif qType == ("INSERT" or "insert" or "Insert"): query = f"INSERT INTO {name} " for i in range(len(col)): if col[i] == col[-1]: query += f"{col[i]}) VALUES " else: query += f"({col[i]}, " for i in range(len(col)): if col[i] == col[-1]: query += "{})" else: query += "({}, " elif qType == ("SELECT" or "select" or "Select"): query = f"SELECT {dataType[1]} FROM {name} WHERE {dataType[0]} == " + "'{}'" elif qType == ("DELETE" or "delete" or "Delete"): query = f"DELETE from {name} WHERE {dataType} = " return query
In python code:
Using the dataframe produced in question#2 and your
class GreenhouseGas(NamedTuple):
Gas: str
Pre_1750: float
Recent: float
Absolute_increase_since_1750: float
Percentage_increase_since_1750: float
def __repr__(self):
return f"Gas: {self.Gas}, Pre-1750: {self.Pre_1750}, Recent: {self.Recent}, Absolute increase since 1750: {self.Absolute_increase_since_1750}, Percentage increase since 1750: {self.Percentage_increase_since_1750}"
def __lt__(self, other):
return self.Recent < other.Recent
def __eq__(self, other):
return self.Gas == other.Gas
@classmethod
def from_list(cls, data):
gas, pre_1750, recent, absolute_increase, percentage_increase = data
return cls(gas, pre_1750, recent, absolute_increase, percentage_increase)
def search_gas(gas_list, gas_name):
for gas in gas_list:
if gas.Gas == gas_name:
return gas
return None
gas_list = [
GreenhouseGas.from_list(["Gas1", 1.0, 2.0, 3.0, 4.0]),
GreenhouseGas.from_list(["Gas2", 5.0, 6.0, 7.0, 8.0]),
GreenhouseGas.from_list(["Gas3", 9.0, 10.0, 11.0, 12.0]),
]
# sort the list by Recent
gas_list.sort()
# search for a specific gas
result = search_gas(gas_list, "Gas2")
print(result)
class Database:
def __init__(self, dbName):
self.dbName = dbName
def connect(self):
global sqliteConnection
try:
sqliteConnection = sqlite3.connect(self.dbName)
cursor = sqliteConnection.cursor()
print("Database created and Successfully Connected to SQLite")
select_Query = "select sqlite_version();"
cursor.execute(select_Query)
record = cursor.fetchall()
except sqlite3.Error as error:
print("Error while connecting to sqlite", error)
def table(self, query):
global sqliteConnection
try:
cursor = sqliteConnection.cursor()
cursor.execute(query)
sqliteConnection.commit()
print("SQLite table created")
except sqlite3.Error as error:
print("Table exists: ", error)
def insert(self, query, df):
global sqliteConnection
try:
cursor = sqliteConnection.cursor()
for row in df.itertuples():
insert_sql = query.format(row[0], row[1])
cursor.execute(insert_sql)
sqliteConnection.commit()
print("Inserted successfully into table")
except sqlite3.Error as error:
print("Failed to insert: ", error)
def search(self, query, value):
global sqliteConnection
cursor = sqliteConnection.cursor()
sel = query.format(value)
cursor.execute(sel)
result = cursor.fetchall()
return result[0][0]
def delete(self, query, id):
global sqliteConnection
try:
cursor = sqliteConnection.cursor()
delete_query = query + str(id)
cursor.execute(delete_query)
sqliteConnection.commit()
print("Record deleted successfully ")
except sqlite3.Error as error:
print("Failed to delete record from sqlite table", error)
def query_builder(self, name, qType, colandType, dataType=[]):
col = list(colandType)
types = list(colandType.values())
if qType == ("TABLE" or "table" or "Table"):
query = f"CREATE TABLE IF NOT EXISTS {name} "
for i in range(len(col)):
if col[i] == col[-1]:
query += f"{col[i]} {types[i]})"
else:
query += f"({col[i]} {types[i]}, "
elif qType == ("INSERT" or "insert" or "Insert"):
query = f"INSERT INTO {name} "
for i in range(len(col)):
if col[i] == col[-1]:
query += f"{col[i]}) VALUES "
else:
query += f"({col[i]}, "
for i in range(len(col)):
if col[i] == col[-1]:
query += "{})"
else:
query += "({}, "
elif qType == ("SELECT" or "select" or "Select"):
query = f"SELECT {dataType[1]} FROM {name} WHERE {dataType[0]} == " + "'{}'"
elif qType == ("DELETE" or "delete" or "Delete"):
query = f"DELETE from {name} WHERE {dataType} = "
return query
Trending now
This is a popular solution!
Step by step
Solved in 4 steps
