我们从Python开源项目中，提取了以下16个代码示例，用于说明如何使用peewee.Proxy()。
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)
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 __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)
