User.py and UserTest.py

database.py

#coding:utf8
__author__ = 'yd'

from sqlalchemy import create_engine,Table
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

#engine = create_engine('sqlite:////tmp/test.db', convert_unicode=True)
_IS_MEM_DB = True
#_IS_MEM_DB = False
if _IS_MEM_DB:
    engine = create_engine('sqlite:///:memory:', echo=False, convert_unicode=True)
else:
    _TESTDB_URL = 'localhost'
    db_string = "postgresql://test:test123@@%s/test_db" % _TESTDB_URL
    engine = create_engine(db_string)

session = scoped_session(sessionmaker(bind=engine))
#autoflush=False,

Base = declarative_base()
Base.query = session.query_property()

def init_db():
    import User
    import Alarm
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)



if __name__ == "__main__":
    init_db()

User.py

#coding:utf8
__author__ = 'yd'
import datetime
from sqlalchemy import or_, Table, ForeignKey
from sqlalchemy import Sequence
from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from database import Base, session


class User(Base):
    __tablename__ = 'USER'
    __table_args__ = {'extend_existing': True}

    id = Column(Integer, Sequence('USER_ID_SEQ'), primary_key=True)
    name = Column(String(10), nullable=False)
    gender = Column(Boolean, )
    birthday = Column(DateTime(10), )
    img = (Integer, )
    shape = (Integer, )
    email = Column(String(30))
    friends = Column(String, default="{}")
    rate = Column(Integer, default=0)
    create_date = Column(DateTime, default=datetime.datetime.utcnow)
    update_date = Column(DateTime, default=datetime.datetime.utcnow)

    def save(self):
        try:
            session.add(self)
            session.commit()
        except Exception as e:
            session.rollback()
            print(e)

        return self

    def remove(self):
        try:
            session.delete(self)
            session.commit()
        except:
            session.rollback()
            raise

    def find(self):
        return session.query(User).filter(or_(User.id == self.id, User.name == self.name)).first()

    def find_alarms(self):
        return session.query(User).filter(User.alarm.any(user_id=self.id)).all()

    def get_data(self):
        info = self.find().__dict__
        return info

    @classmethod
    def update(self, _id, name=name, gender=gender):
        user = session.query(User).filter(User.id == _id).first()
        if user:
            user.name = name
            user.gender = gender
            user.save()

    @classmethod
    def delete(self, user):
        session.delete(user)

    @classmethod
    def get_list(self):
        query = session.query(User)
        data_list = []
        for each in query.all():
            info = each.__dict__
            data_list.append(info)
        return {'data':data_list}
        # return session.query(User).all()

if __name__ == "__main__":
    pass

UserTest.py

#coding:utf8
import unittest
from User import User
from Alarm import Alarm
from database import init_db

__author__ = 'yd'


class UserTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        init_db()

    def test_save(self):
        user = User(name="test")
        user.save()

        user = user.find()
        self.assertEqual(user.name, "test")

    def test_find(self):
        user = User(name="test").save().find()
        self.assertEqual(user.name, "test")

    def test_get_list(self):
        init_db()
        user = User(name="test")
        user1 = User(name="test2")
        user.save()
        user1.save()

        user_list = []

        user_list.append(user.get_data())
        user_list.append(user1.get_data())

        self.assertEqual(User.get_list()['data'], user_list)

    def test_update(self):
        init_db()
        user = User(name="test", gender=0)
        user.save()

        User.update(user.id, name="test_change", gender=1)

        get_user = User(id=user.id).find()

        self.assertEqual(get_user.name, "test_change")
        self.assertEqual(get_user.gender, 1)

    def test_delete(self):
        init_db()
        user = User(name="test", gender=0)
        user.save()
        user.remove()

        self.assertTrue(User(name="test").find() == None)

if __name__ == '__main__':
    unittest.main()