from collections import OrderedDict
from flask import json, jsonify
from sqlalchemy import func, text
from sqlalchemy import Column, Boolean, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declared_attr
from app import db
class CoreModel(db.Model):
__abstract__ = True
# __tablename__ = name_model as name
@declared_attr
def __tablename__(cls):
cn = cls.__name__.lower()
_i = cn.rfind('_model')
if _i > 0:
return cn[:_i]
else:
return cn
# result as list
@classmethod
def to_list(cls, query_result, obj_field_names=None):
data_array = []
for result_item in query_result:
item = result_item.as_dict()
if obj_field_names is not None:
for obj_field_name in obj_field_names:
obj_field = getattr(result_item, obj_field_name)
if isinstance(obj_field, CoreModel):
item[obj_field_name] = obj_field.as_dict()
data_array.append(item)
return data_array
# model as dict
def as_dict(self, obj_field_names=None):
data_object = OrderedDict()
for key in self.__mapper__.c.keys():
val_field = getattr(self, key)
if not isinstance(val_field, CoreModel):
data_object[key] = val_field
if obj_field_names is not None:
for obj_field_name in obj_field_names:
obj_field = getattr(self, obj_field_name)
if isinstance(obj_field, CoreModel):
data_object[obj_field_name] = obj_field.as_dict()
return data_object
def addme(self):
db.session.add(self)
return self
def save(self):
db.session.add(self)
db.session.commit()
return self
def rollback(self):
db.session.rollback()
return self
@classmethod
def commit(cls):
db.session.commit()
@classmethod
def execute(cls, sql, params=None):
if params is None:
params = {}
resultproxy = db.engine.execute(text(sql), params)
d, a = {}, []
for rowproxy in resultproxy:
# rowproxy.items() returns an array like [(key0, value0), (key1, value1)]
for tup in rowproxy.items():
# build up the dictionary
d = {**d, **{tup[0]: tup[1]}}
a.append(d)
return a
class BaseModel(CoreModel):
__abstract__ = True
id = Column(Integer, primary_key=True)
created_at = Column(DateTime, server_default=func.now())
modified_at = Column(DateTime, server_default=func.now(),
server_onupdate=func.now())
deleted = Column(Boolean, nullable=False, default=False)
@classmethod
def get_first(self):
return self.query.order_by(self.id.asc()).first()
@classmethod
def get_last(self):
return self.query.order_by(self.id.desc()).first()
@classmethod
def get_list_all(self):
lst = []
for item in self.query.order_by(self.id.asc()).all():
lst.append(item.as_dict())
return lst
@classmethod
def get_list(self, **kwargs):
query = self.query.filter_by(**kwargs)
lst = []
for item in query.order_by(self.id.asc()).all():
lst.append(item.as_dict())
return lst
class FullBaseModel(BaseModel):
__abstract__ = True
created_ip = Column(String(length=50))
created_by = Column(Integer, ForeignKey('user.id'))
modified_ip = Column(String(length=50))
modified_by = Column(Integer, ForeignKey('user.id'))
ordering = Column(Integer, nullable=False, default=0)