def __init__(self, engine_cls, instance, *args, **kwargs):
"""
Initiate database.
:param engine_cls: Engine class
:param instance: Instance of the app.
:param args: *
:param kwargs: **
:type instance: pyplanet.core.instance.Instance
"""
self.engine = engine_cls(*args, **kwargs)
self.instance = instance
self.migrator = Migrator(self.instance, self)
self.registry = Registry(self.instance, self)
self.objects = peewee_async.Manager(self.engine, loop=self.instance.loop)
# Don't allow any sync code.
if hasattr(self.engine, 'allow_sync'):
self.engine.allow_sync = False
Proxy.initialize(self.engine)
python类Proxy()的实例源码
def build_fake_model(self, name):
"""
Build a fake model with some defaults and the given table name.
We need this so we can perform operations that actually require a model class.
:param name: Name of database table.
:return: A new model class.
:rtype: peewee.Model
"""
class FakeModel(peewee.Model):
class Meta:
database = peewee.Proxy()
primary_key = False
indexes = []
constraints = []
db_table = name
return FakeModel
def add_to_class(self, model_class, name):
if isinstance(self._through_model, Proxy):
def callback(through_model):
self._through_model = through_model
self.add_to_class(model_class, name)
self._through_model.attach_callback(callback)
return
elif isinstance(self._through_model, DeferredThroughModel):
self._through_model.set_field(model_class, self, name)
return
self.name = name
self.model_class = model_class
if not self.verbose_name:
self.verbose_name = re.sub('_+', ' ', name).title()
setattr(model_class, name, self._get_descriptor())
if not self._is_backref:
backref = AioManyToManyField(
self.model_class,
through_model=self._through_model,
_is_backref=True)
related_name = self._related_name or model_class._meta.name + 's'
backref.add_to_class(self.rel_model, related_name)
def db(self) -> dict:
"""
Sets up the database proxy and exposes the database variables in a `db` property.
"""
# Create a database proxy (placeholder) to be filled at runtime with the actual database object.
self._config_dict['aryas']['db']['_db_proxy'] = Proxy()
return self._config_dict['aryas']['db']
def foreign_key(self, coltype, name, references, **kwargs):
"""
Add a foreign key to the model.
This has some special cases, which is why it's not handled like all the other column types.
:param name: Name of the foreign key.
:param references: Table name in the format of "table.column" or just
"table" (and id will be default column).
:param kwargs: Additional kwargs to pass to the column instance.
You can also provide "on_delete" and "on_update" to add constraints.
:return: None
"""
try:
rel_table, rel_column = references.split('.', 1)
except ValueError:
rel_table, rel_column = references, 'id'
# Create a dummy model that we can relate this field to.
# Add the foreign key as a local field on the dummy model.
# We only do this so that Peewee can generate the nice foreign key constraint for us.
class DummyRelated(peewee.Model):
class Meta:
primary_key = False
database = peewee.Proxy()
db_table = rel_table
rel_field_class = FIELD_TO_PEEWEE.get(coltype, peewee.IntegerField)
rel_field = rel_field_class()
rel_field.add_to_class(DummyRelated, rel_column)
field = peewee.ForeignKeyField(DummyRelated, db_column=name, to_field=rel_column, **kwargs)
field.add_to_class(self.model, name)
def load_database(self, database):
"""
Load the given database, whatever it might be.
A connection string: ``sqlite:///database.sqlite``
A dictionary: ``{'engine': 'SqliteDatabase', 'name': 'database.sqlite'}``
A peewee.Database instance: ``peewee.SqliteDatabase('database.sqlite')``
:param database: Connection string, dict, or peewee.Database instance to use.
:raises: peewee.DatabaseError if database connection cannot be established.
:return: Database connection.
:rtype: peewee.Database instance.
"""
# It could be an actual instance...
if isinstance(database, (peewee.Proxy, peewee.Database)):
return database
# It could be a dictionary...
if isinstance(database, dict):
try:
name = database.pop('name')
engine = database.pop('engine')
except KeyError:
error_msg = 'Configuration dict must specify "name" and "engine" keys.'
raise peewee.DatabaseError(error_msg)
db_class = pydoc.locate(engine)
if not db_class:
raise peewee.DatabaseError('Unable to import engine class: {}'.format(engine))
return db_class(name, **database)
# Or it could be a database URL.
return url_connect(database)
def _load_database(self, app):
config_value = app.config['DATABASE']
if isinstance(config_value, dict):
database = self._load_from_config_dict(dict(config_value))
else:
# Assume a database connection URL.
database = db_url_connect(config_value)
if isinstance(self.database, Proxy):
self.database.initialize(database)
else:
self.database = database
def Model(self):
if self._app is None:
database = getattr(self, 'database', None)
if database is None:
self.database = Proxy()
if not hasattr(self, '_model_class'):
self._model_class = self.get_model_class()
return self._model_class
def _load_database(self, app, config_value):
if isinstance(config_value, Database):
database = config_value
elif isinstance(config_value, dict):
database = self._load_from_config_dict(dict(config_value))
else:
# Assume a database connection URL.
database = db_url_connect(config_value)
if isinstance(self.database, Proxy):
self.database.initialize(database)
else:
self.database = database
def Model(self):
if self._app is None:
database = getattr(self, 'database', None)
if database is None:
self.database = Proxy()
if not hasattr(self, '_model_class'):
self._model_class = self.get_model_class()
return self._model_class
def _load_database(self, app, config_value):
if isinstance(config_value, Database):
database = config_value
elif isinstance(config_value, dict):
database = self._load_from_config_dict(dict(config_value))
else:
# Assume a database connection URL.
database = db_url_connect(config_value)
if isinstance(self.database, Proxy):
self.database.initialize(database)
else:
self.database = database
def Model(self):
if self._app is None:
database = getattr(self, 'database', None)
if database is None:
self.database = Proxy()
if not hasattr(self, '_model_class'):
self._model_class = self.get_model_class()
return self._model_class
def _load_database(self, app, config_value):
if isinstance(config_value, Database):
database = config_value
elif isinstance(config_value, dict):
database = self._load_from_config_dict(dict(config_value))
else:
# Assume a database connection URL.
database = db_url_connect(config_value)
if isinstance(self.database, Proxy):
self.database.initialize(database)
else:
self.database = database
def Model(self):
if self._app is None:
database = getattr(self, 'database', None)
if database is None:
self.database = Proxy()
if not hasattr(self, '_model_class'):
self._model_class = self.get_model_class()
return self._model_class
def __init__(self, app=None):
"""Initialize the plugin."""
self.app = app
self.database = pw.Proxy()
if app is not None:
self.init_app(app)
def test_get_model_supported(version):
M = model.get_model(version)
assert M
assert M.database_proxy
assert isinstance(M.database_proxy, peewee.Proxy)
