python alembic sqlalchemy base model __init__

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)