def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
python类declarative_base()的实例源码
def init():
config = configparser.ConfigParser()
config.read('lama/conf/project.conf')
# get informations for connexion
try:
host = config["DATABASE"]["host"]
database = config["DATABASE"]["database"]
user = config["DATABASE"]["user"]
password = config["DATABASE"]["password"]
except KeyError as e:
logging.error("Error project.conf[DATABASE] : {} missing.".format(str(e)))
exit(1)
Lamadb.Base = declarative_base()
connect = "postgresql://{}:{}@{}/{}".format(user, password, host, database)
Lamadb.engine = create_engine(connect)
Lamadb.conn = Lamadb.engine.connect()
def make_sqlalchemy_redis_base(
name='ModelSQLAlchemyRedisBase',
bind=None, metadata=None,
mapper=None, key_separator=None,
metaclass=ModelSQLAlchemyRedisBaseMeta,
cls=ModelSQLAlchemyRedisBaseSuper,
constructor=ModelSQLAlchemyRedisBaseSuper.__init__):
base = declarative_base(
name=name, metaclass=metaclass,
cls=cls, bind=bind, metadata=metadata,
mapper=mapper, constructor=constructor)
if key_separator is not None:
base.__key_separator__ = key_separator
return base
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def connect(self):
self.build_connection_string()
if self.m_connection_str == None:
self.lg("Not connecting to this database", 0)
return None
self.lg("Connecting to databases(" + str(self.m_connection_str) + ") Autocommit(" + str(self.m_autocommit) + ") Autoflush(" + str(self.m_autoflush) + ")", 7)
Base = declarative_base()
self.m_engine = create_engine(self.m_connection_str,
echo=False)
self.m_connection = self.m_engine.connect()
self.m_session = scoped_session(sessionmaker(autocommit = self.m_autocommit,
autoflush = self.m_autoflush,
bind = self.m_engine))
self.lg("Connected to DB(" + str(self.m_name) + ") DBTables(" + str(self.m_database_name) + ")", 7)
return None
# end of connect
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def __init__(self):
super(AlchemyBase, self).__init__()
def fk_fixed_width(constraint, table):
str_tokens = [table.name] +\
[element.parent.name for element in constraint.elements] +\
[element.target_fullname for element in constraint.elements]
guid = uuid.uuid5(uuid.NAMESPACE_OID, "_".join(str_tokens).encode('ascii'))
return str(guid)
convention = {
"fk_fixed_width": fk_fixed_width,
"ix": 'ix_%(column_0_label)s',
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(column_0_name)s",
"fk": "fk_%(fk_fixed_width)s",
"pk": "pk_%(table_name)s"
}
metadata = MetaData(naming_convention=convention)
self.Model = declarative_base(metadata=metadata, cls=Model, name='Model', metaclass=_BoundDeclarativeMeta)
self.Model.query = _QueryProperty(self)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables:
cls.metadata.create_all(config.db)
def __init__(self, data_path, models, encoding=None, markdown_config=None):
"""Constructor.
Args:
data_path: The full path to where the database files can be found.
models: Loaded model/field data.
encoding: The encoding to load files as ('utf-8', etc). If 'None', will
default to the system-preferred default encoding
"""
self.encoding = encoding
self.tables = dict()
self.data_path = data_path
self.models = models
self.markdown_config = markdown_config
self.engine = create_engine('sqlite:///:memory:')
self.Base = declarative_base()
self.session = sessionmaker(bind=self.engine)()
set_global('session', self.session)
self.find_backrefs()
self.create_db(models)
def test_no_inflect(self):
Table(
'simple_items', self.metadata,
Column('id', INTEGER, primary_key=True)
)
assert self.generate_code(noinflect=True) == """\
# coding: utf-8
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
class SimpleItems(Base):
__tablename__ = 'simple_items'
id = Column(Integer, primary_key=True)
"""
def test_table_kwargs(self):
Table(
'simple_items', self.metadata,
Column('id', INTEGER, primary_key=True),
schema='testschema'
)
assert self.generate_code() == """\
# coding: utf-8
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
class SimpleItem(Base):
__tablename__ = 'simple_items'
__table_args__ = {'schema': 'testschema'}
id = Column(Integer, primary_key=True)
"""
def test_pascal(self):
Table(
'CustomerAPIPreference', self.metadata,
Column('id', INTEGER, primary_key=True)
)
assert self.generate_code() == """\
# coding: utf-8
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
class CustomerAPIPreference(Base):
__tablename__ = 'CustomerAPIPreference'
id = Column(Integer, primary_key=True)
"""
def test_underscore(self):
Table(
'customer_api_preference', self.metadata,
Column('id', INTEGER, primary_key=True)
)
assert self.generate_code() == """\
# coding: utf-8
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
class CustomerApiPreference(Base):
__tablename__ = 'customer_api_preference'
id = Column(Integer, primary_key=True)
"""
def test_pascal_underscore(self):
Table(
'customer_API_Preference', self.metadata,
Column('id', INTEGER, primary_key=True)
)
assert self.generate_code() == """\
# coding: utf-8
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
metadata = Base.metadata
class CustomerAPIPreference(Base):
__tablename__ = 'customer_API_Preference'
id = Column(Integer, primary_key=True)
"""
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def connect(self):
self.build_connection_string()
if self.m_connection_str == None:
self.lg("Not connecting to this database", 0)
return None
self.lg("Connecting to databases(" + str(self.m_connection_str) + ") Autocommit(" + str(self.m_autocommit) + ") Autoflush(" + str(self.m_autoflush) + ")", 7)
Base = declarative_base()
self.m_engine = create_engine(self.m_connection_str,
echo=False)
self.m_connection = self.m_engine.connect()
self.m_session = scoped_session(sessionmaker(autocommit = self.m_autocommit,
autoflush = self.m_autoflush,
bind = self.m_engine))
self.lg("Connected to DB(" + str(self.m_name) + ") DBTables(" + str(self.m_database_name) + ")", 7)
return None
# end of connect
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
def _create_schema(self):
log.info("Creating schema.")
if self._schema:
tmp_Base = declarative_base()
d = _init_daos(tmp_Base, DObject())
for t in tmp_Base.metadata.tables.values():
t.schema = self.schema
try:
tmp_Base.metadata.create_all(self.engine)
except ProgrammingError:
with self.open_session() as sess:
st = "CREATE SCHEMA {}".format(self.schema)
sess.execute(st)
sess.commit()
tmp_Base.metadata.create_all(self.engine)
else:
Base.metadata.create_all(self.engine)
test_sql.py 文件源码
项目：PyDataLondon29-EmbarrassinglyParallelDAWithAWSLambda
作者: SignalMedia
项目源码
文件源码
阅读 32
收藏 0
点赞 0
评论 0
def test_temporary_table(self):
test_data = u'Hello, World!'
expected = DataFrame({'spam': [test_data]})
Base = declarative.declarative_base()
class Temporary(Base):
__tablename__ = 'temp_test'
__table_args__ = {'prefixes': ['TEMPORARY']}
id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
spam = sqlalchemy.Column(sqlalchemy.Unicode(30), nullable=False)
Session = sa_session.sessionmaker(bind=self.conn)
session = Session()
with session.transaction:
conn = session.connection()
Temporary.__table__.create(conn)
session.add(Temporary(spam=test_data))
session.flush()
df = sql.read_sql_query(
sql=sqlalchemy.select([Temporary.spam]),
con=conn,
)
tm.assert_frame_equal(df, expected)
def connect(app):
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.ext.declarative import declarative_base
from generic_social_network.app.models.tables.my_model import MyModel
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + _db_file_location()
class MySQLAlchemy(SQLAlchemy):
def make_declarative_base(self):
from flask.ext.sqlalchemy import _BoundDeclarativeMeta, _QueryProperty
base = declarative_base(cls=MyModel, name='MyModel', metaclass=_BoundDeclarativeMeta)
base.query = _QueryProperty(self)
return base
db = MySQLAlchemy(app)
# db.engine.echo = True
return db
def gen_relation_models():
Base = declarative_base()
class RelationTestModel(Base):
__tablename__ = "sanity_check_test_2"
id = Column(Integer, primary_key=True)
class RelationTestModel2(Base):
__tablename__ = "sanity_check_test_3"
id = Column(Integer, primary_key=True)
test_relationship_id = Column(ForeignKey("sanity_check_test_2.id"))
test_relationship = relationship(RelationTestModel, primaryjoin=test_relationship_id == RelationTestModel.id)
return Base, RelationTestModel, RelationTestModel2
def gen_declarative():
Base = declarative_base()
class DeclarativeTestModel(Base):
__tablename__ = "sanity_check_test_4"
id = Column(Integer, primary_key=True)
@declared_attr
def _password(self):
return Column('password', String(256), nullable=False)
@hybrid_property
def password(self):
return self._password
return Base, DeclarativeTestModel
def sa_model():
from sqlalchemy import create_engine, Table, Column, Integer
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///:memory:')
Base = declarative_base()
class Test(Base):
__tablename__ = 'test'
a = Column(Integer, primary_key=True)
b = Column(Integer)
c = Column(Integer)
d = Column(Integer)
Base.metadata.create_all(engine)
return Test
def _with_register_classes(cls, fn):
cls_registry = cls.classes
class FindFixtureDeclarative(DeclarativeMeta):
def __init__(cls, classname, bases, dict_):
cls_registry[classname] = cls
return DeclarativeMeta.__init__(
cls, classname, bases, dict_)
class DeclarativeBasic(object):
__table_cls__ = schema.Table
_DeclBase = declarative_base(metadata=cls.metadata,
metaclass=FindFixtureDeclarative,
cls=DeclarativeBasic)
cls.DeclarativeBasic = _DeclBase
fn()
if cls.metadata.tables and cls.run_create_tables:
cls.metadata.create_all(config.db)
