namedtuple搭配sqlalchemy进行测试

sqlalchemy_test.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Author: ficapy
# Create: '5/25/17'

import inspect
from datetime import datetime
from sqlalchemy import inspection
from collections import namedtuple
from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

sqlalchemy_db = 'sqlite://'

Base = declarative_base()


class Demo(Base):
    __tablename__ = 'demo'
    id = Column(Integer, primary_key=True)
    field1 = Column(Integer)
    field2 = Column(String)
    field3 = Column(DateTime)


uline_engine = create_engine(sqlalchemy_db, pool_recycle=3600, echo=False)

Base.metadata.bind = uline_engine
Base.metadata.create_all(checkfirst=True)
db_Session = sessionmaker()
db_Session.configure(bind=uline_engine)

session = db_Session()

task = namedtuple('demo', ['field1', 'field2', 'field3'])


def __new__(cls, **kwargs):
    name = cls.__name__
    for c in Base._decl_class_registry.values():
        if hasattr(c, '__tablename__') and c.__tablename__ == name:
            field_type_map = {column.key: column.columns[0].type.python_type for column in
                              inspection.inspect(c).mapper.column_attrs}

    ret = []
    for i in cls._fields:
        if i in kwargs:
            ret.append(kwargs[i])
        else:
            if isinstance(1, field_type_map[i]):
                ret.append(2)
            elif isinstance('1', field_type_map[i]):
                ret.append('')
            elif isinstance(datetime.now(), field_type_map[i]):
                ret.append(datetime.now())
            else:
                raise Exception('{}:{}'.format(i, field_type_map[i]))
    return tuple.__new__(cls, ret)


for i in [i for i in inspect.currentframe().f_globals.values()]:
    if hasattr(i, '_fields'):
        i.__new__ = __new__

task1 = task()


def save_db(data):
    name = type(data).__name__
    for c in Base._decl_class_registry.values():
        if hasattr(c, '__tablename__') and c.__tablename__ == name:
            session.add(c(**data._asdict()))
            session.commit()


save_db(task1)

ret = session.query(Demo).first()
print({c.name: getattr(ret, c.name, None) for c in ret.__table__.columns})