我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.ext.declarative.declarative_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 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 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 __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: 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 _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)
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 _get_declarative_base(): try: if FORCE_AVOID_INTERNAL_PACKAGES: # TODO(DATAPIPE-1506|abrar): Currently we have # force_avoid_internal_packages as a means of simulating an absence # of a yelp's internal package. And all references # of force_avoid_internal_packages have to be removed from # schematizer after we have completely ready for open source. raise ImportError from yelp_conn.session import declarative_base except ImportError: from sqlalchemy.ext.declarative import declarative_base return declarative_base() # The common declarative base used by every data model.
def bootstrap_events(f): @wraps(f) def wrapped(*args, **kwargs): global _USER_BASE if f.__name__ == "declarative_base": base = f(*args, **kwargs) base.metadata.create_all = bootstrap_events(base.metadata.create_all) _USER_BASE = base return base elif f.__name__ == "create_all": tables = [v for k, v in _USER_BASE.metadata.tables.items()] for table in tables: if is_zdb_table(table): before_create(table) after_create(table) f(*args, **kwargs) else: return f(*args, **kwargs) return wrapped
def get(self, request,app=None): pycsw_settings = build_pycsw_settings(app) server = Csw(rtconfig=pycsw_settings, env=request.META.copy()) if not app: app = "all" #request by named app, use app related view record_table = Application.get_view_name(app) try: if not self.application_records.get(app,None): base = declarative_base(bind=server.repository.engine,mapper=Mapper) self.application_records[app] = type('dataset', (base,), dict(__tablename__=record_table,__table_args__={'autoload': True,'schema': None},__mapper_args__={"primary_key":["id"]})) server.repository.dataset = self.application_records[app] except: pass server.request = "http://{}{}".format(get_current_site(request), reverse("csw_endpoint")) server.requesttype = request.method server.kvp = self._normalize_params(request.GET) response = server.dispatch() return HttpResponse(response, content_type="application/xml")
def __init__(self,SQLALCHEMY_DATABASE_URI=None,database_name=None,db_uri=None): #SQLALCHEMY_DATABASE_URI = sql_uri #self.config_uri = db_uri #self.engine = create_engine(SQLALCHEMY_DATABASE_URI) #self.database_name = database_name config_uri = config.DB_URI self.engine = create_engine(config.SQLALCHEMY_DATABASE_URI) self.database_name = config.database_name self.Base = declarative_base() self.metadata = MetaData(bind=self.engine) self.auto_map = automap_base()
def as_declarative(**kw): """ Class decorator for :func:`.declarative_base`. Provides a syntactical shortcut to the ``cls`` argument sent to :func:`.declarative_base`, allowing the base class to be converted in-place to a "declarative" base:: from sqlalchemy.ext.declarative import as_declarative @as_declarative() class Base(object): @declared_attr def __tablename__(cls): return cls.__name__.lower() id = Column(Integer, primary_key=True) class MyMappedClass(Base): # ... All keyword arguments passed to :func:`.as_declarative` are passed along to :func:`.declarative_base`. .. versionadded:: 0.8.3 .. seealso:: :func:`.declarative_base` """ def decorate(cls): kw['cls'] = cls kw['name'] = cls.__name__ return declarative_base(**kw) return decorate
def make_declarative_base(self, metadata=None): """Creates the declarative base.""" base = declarative_base(cls=Model, name='Model', metadata=metadata, metaclass=_BoundDeclarativeMeta) base.query = _QueryProperty(self) return base
def setUp(self): self.Base = declarative_base() self.engine = create_engine(self.db_uri, echo=False)
