craftbeerpi4-pione/core/database/orm_framework.py

157 lines
5 KiB
Python
Raw Normal View History

2018-11-01 21:25:42 +01:00
import json
import aiosqlite
2018-11-16 20:35:59 +01:00
import os
2018-11-01 21:25:42 +01:00
# Thats the name of the database file
TEST_DB = "./craftbeerpi.db"
class DBModel(object):
__priamry_key__ = "id"
__as_array__ = False
__order_by__ = None
__json_fields__ = []
def __init__(self, args):
2018-11-16 20:35:59 +01:00
2018-11-01 21:25:42 +01:00
self.__setattr__(self.__priamry_key__, args[self.__priamry_key__])
for f in self.__fields__:
2018-11-16 20:35:59 +01:00
2018-11-01 21:25:42 +01:00
if f in self.__json_fields__:
2018-11-16 20:35:59 +01:00
if args.get(f) is not None:
2018-11-01 21:25:42 +01:00
if isinstance(args[f], dict) or isinstance(args[f], list):
2018-11-16 20:35:59 +01:00
self.__setattr__(f, args.get(f))
2018-11-01 21:25:42 +01:00
else:
2018-11-16 20:35:59 +01:00
self.__setattr__(f, json.loads(args.get(f, "{}")))
2018-11-01 21:25:42 +01:00
else:
self.__setattr__(f, None)
else:
2018-11-16 20:35:59 +01:00
self.__setattr__(f, args.get(f))
2018-11-01 21:25:42 +01:00
@classmethod
async def test_connection(self):
2018-11-04 00:47:26 +01:00
2018-11-01 21:25:42 +01:00
async with aiosqlite.connect(TEST_DB) as db:
assert isinstance(db, aiosqlite.Connection)
2018-11-16 20:35:59 +01:00
this_directory = os.path.dirname(__file__)
qry = open(os.path.join(this_directory, '../sql/create_table_user.sql'), 'r').read()
2018-11-01 21:25:42 +01:00
cursor = await db.executescript(qry)
@classmethod
async def get_all(cls):
2018-11-04 00:47:26 +01:00
2018-11-01 21:25:42 +01:00
if cls.__as_array__ is True:
result = []
else:
result = {}
async with aiosqlite.connect(TEST_DB) as db:
if cls.__order_by__ is not None:
sql = "SELECT * FROM %s ORDER BY %s.'%s'" % (cls.__table_name__, cls.__table_name__, cls.__order_by__)
else:
sql = "SELECT * FROM %s" % cls.__table_name__
2018-11-16 20:35:59 +01:00
db.row_factory = DBModel.dict_factory
2018-11-01 21:25:42 +01:00
async with db.execute(sql) as cursor:
async for row in cursor:
2018-11-16 20:35:59 +01:00
2018-11-01 21:25:42 +01:00
if cls.__as_array__ is True:
result.append(cls(row))
else:
2018-11-16 20:35:59 +01:00
result[row.get("id")] = cls(row)
2018-11-01 21:25:42 +01:00
await cursor.close()
return result
@classmethod
async def get_one(cls, id):
async with aiosqlite.connect(TEST_DB) as db:
db.row_factory = aiosqlite.Row
2018-11-16 20:35:59 +01:00
db.row_factory = DBModel.dict_factory
2018-11-01 21:25:42 +01:00
async with db.execute("SELECT * FROM %s WHERE %s = ?" % (cls.__table_name__, cls.__priamry_key__), (id,)) as cursor:
row = await cursor.fetchone()
if row is not None:
return cls(row)
else:
return None
@classmethod
async def delete(cls, id):
async with aiosqlite.connect(TEST_DB) as db:
await db.execute("DELETE FROM %s WHERE %s = ? " % (cls.__table_name__, cls.__priamry_key__), (id,))
await db.commit()
@classmethod
async def insert(cls, **kwargs):
async with aiosqlite.connect(TEST_DB) as db:
if cls.__priamry_key__ is not None and cls.__priamry_key__ in kwargs:
query = "INSERT INTO %s (%s, %s) VALUES (?, %s)" % (
cls.__table_name__,
cls.__priamry_key__,
', '.join("'%s'" % str(x) for x in cls.__fields__),
', '.join(['?'] * len(cls.__fields__)))
data = ()
data = data + (kwargs.get(cls.__priamry_key__),)
for f in cls.__fields__:
if f in cls.__json_fields__:
data = data + (json.dumps(kwargs.get(f)),)
else:
data = data + (kwargs.get(f),)
else:
query = 'INSERT INTO %s (%s) VALUES (%s)' % (
cls.__table_name__,
', '.join("'%s'" % str(x) for x in cls.__fields__),
', '.join(['?'] * len(cls.__fields__)))
data = ()
for f in cls.__fields__:
if f in cls.__json_fields__:
data = data + (json.dumps(kwargs.get(f)),)
else:
data = data + (kwargs.get(f),)
2018-11-04 00:47:26 +01:00
2018-11-01 21:25:42 +01:00
cursor = await db.execute(query, data)
await db.commit()
i = cursor.lastrowid
kwargs["id"] = i
return cls(kwargs)
@classmethod
async def update(cls, **kwargs):
async with aiosqlite.connect(TEST_DB) as db:
query = 'UPDATE %s SET %s WHERE %s = ?' % (cls.__table_name__, ', '.join("'%s' = ?" % str(x) for x in cls.__fields__), cls.__priamry_key__)
data = ()
for f in cls.__fields__:
if f in cls.__json_fields__:
data = data + (json.dumps(kwargs.get(f)),)
else:
data = data + (kwargs.get(f),)
data = data + (kwargs.get(cls.__priamry_key__),)
cursor = await db.execute(query, data)
await db.commit()
return cls(kwargs)
2018-11-16 20:35:59 +01:00
@classmethod
def dict_factory(cls, cursor, row):
d = {}
for idx, col in enumerate(cursor.description):
d[col[0]] = row[idx]
return d
2018-11-01 21:25:42 +01:00