我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlalchemy.dialects.postgresql.dialect()。
def create(self, bind=None, checkfirst=True): """Emit ``CREATE TYPE`` for this :class:`~.postgresql.ENUM`. If the underlying dialect does not support Postgresql CREATE TYPE, no action is taken. :param bind: a connectable :class:`.Engine`, :class:`.Connection`, or similar object to emit SQL. :param checkfirst: if ``True``, a query against the PG catalog will be first performed to see if the type does not exist already before creating. """ if not bind.dialect.supports_native_enum: return if not checkfirst or \ not bind.dialect.has_type( bind, self.name, schema=self.schema): bind.execute(CreateEnumType(self))
def create(self, bind=None, checkfirst=True): """Emit ``CREATE TYPE`` for this :class:`~.postgresql.ENUM`. If the underlying dialect does not support Postgresql CREATE TYPE, no action is taken. :param bind: a connectable :class:`.Engine`, :class:`.Connection`, or similar object to emit SQL. :param checkfirst: if ``True``, a query against the PG catalog will be first performed to see if the type does not exist already before creating. """ if not bind.dialect.supports_native_enum: return if not checkfirst or \ not bind.dialect.has_type(bind, self.name, schema=self.schema): bind.execute(CreateEnumType(self))
def get_column_specification(self, column, **kwargs): colspec = self.preparer.format_column(column) impl_type = column.type.dialect_impl(self.dialect) if column.primary_key and \ column is column.table._autoincrement_column and \ not isinstance(impl_type, sqltypes.SmallInteger) and \ ( column.default is None or ( isinstance(column.default, schema.Sequence) and column.default.optional )): if isinstance(impl_type, sqltypes.BigInteger): colspec += " BIGSERIAL" else: colspec += " SERIAL" else: colspec += " " + self.dialect.type_compiler.process(column.type) default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default if not column.nullable: colspec += " NOT NULL" return colspec
def drop(self, bind=None, checkfirst=True): """Emit ``DROP TYPE`` for this :class:`~.postgresql.ENUM`. If the underlying dialect does not support Postgresql DROP TYPE, no action is taken. :param bind: a connectable :class:`.Engine`, :class:`.Connection`, or similar object to emit SQL. :param checkfirst: if ``True``, a query against the PG catalog will be first performed to see if the type actually exists before dropping. """ if not bind.dialect.supports_native_enum: return if not checkfirst or \ bind.dialect.has_type(bind, self.name, schema=self.schema): bind.execute(DropEnumType(self))
def get_enums(self, schema=None): """Return a list of ENUM objects. Each member is a dictionary containing these fields: * name - name of the enum * schema - the schema name for the enum. * visible - boolean, whether or not this enum is visible in the default search path. * labels - a list of string labels that apply to the enum. :param schema: schema name. If None, the default schema (typically 'public') is used. May also be set to '*' to indicate load enums for all schemas. .. versionadded:: 1.0.0 """ schema = schema or self.default_schema_name return self.dialect._load_enums(self.bind, schema)
def render_sql(db, q, inline=False): """ Render the sql used by a query (only works for Postgres) :param q (query): an SQLAlchemy query object :param inline (bool): inline parameters? :return: str """ compiled_statement = q.statement.compile(dialect=postgresql.dialect()) pretty_statement = sql_format(str(compiled_statement), reindent=True) if inline: with db.session.connection().connection.connection.cursor() as cur: return cur.mogrify(pretty_statement, compiled_statement.params).decode('utf-8') else: return pretty_statement + ("\nparameters: {}".format(str(compiled_statement.params)) if compiled_statement.params else '')
def bind_processor(self, dialect): if self.as_uuid: def process(value): if value is not None: value = util.text_type(value) return value return process else: return None
def result_processor(self, dialect, coltype): if self.as_uuid: def process(value): if value is not None: value = _python_UUID(value) return value return process else: return None
def bind_processor(self, dialect): item_proc = self.item_type.\ dialect_impl(dialect).\ bind_processor(dialect) def process(value): if value is None: return value else: return self._proc_array( value, item_proc, self.dimensions, list) return process
def result_processor(self, dialect, coltype): item_proc = self.item_type.\ dialect_impl(dialect).\ result_processor(dialect, coltype) def process(value): if value is None: return value else: return self._proc_array( value, item_proc, self.dimensions, tuple if self.as_tuple else list) return process
def render_literal_value(self, value, type_): value = super(PGCompiler, self).render_literal_value(value, type_) if self.dialect._backslash_escapes: value = value.replace('\\', '\\\\') return value
def get_column_specification(self, column, **kwargs): colspec = self.preparer.format_column(column) impl_type = column.type.dialect_impl(self.dialect) if column.primary_key and \ column is column.table._autoincrement_column and \ ( self.dialect.supports_smallserial or not isinstance(impl_type, sqltypes.SmallInteger) ) and ( column.default is None or ( isinstance(column.default, schema.Sequence) and column.default.optional )): if isinstance(impl_type, sqltypes.BigInteger): colspec += " BIGSERIAL" elif isinstance(impl_type, sqltypes.SmallInteger): colspec += " SMALLSERIAL" else: colspec += " SERIAL" else: colspec += " " + self.dialect.type_compiler.process(column.type, type_expression=column) default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default if not column.nullable: colspec += " NOT NULL" return colspec
def visit_enum(self, type_, **kw): if not type_.native_enum or not self.dialect.supports_native_enum: return super(PGTypeCompiler, self).visit_enum(type_, **kw) else: return self.visit_ENUM(type_, **kw)
def visit_ENUM(self, type_, **kw): return self.dialect.identifier_preparer.format_type(type_)
def get_table_oid(self, table_name, schema=None): """Return the OID for the given table name.""" return self.dialect.get_table_oid(self.bind, table_name, schema, info_cache=self.info_cache)
def get_foreign_table_names(self, schema=None): """Return a list of FOREIGN TABLE names. Behavior is similar to that of :meth:`.Inspector.get_table_names`, except that the list is limited to those tables tha report a ``relkind`` value of ``f``. .. versionadded:: 1.0.0 """ schema = schema or self.default_schema_name return self.dialect._get_foreign_table_names(self.bind, schema)
def fire_sequence(self, seq, type_): return self._execute_scalar(( "select nextval('%s')" % self.dialect.identifier_preparer.format_sequence(seq)), type_)
def get_column_specification(self, column, **kwargs): colspec = self.preparer.format_column(column) impl_type = column.type.dialect_impl(self.dialect) if column.primary_key and \ column is column.table._autoincrement_column and \ ( self.dialect.supports_smallserial or not isinstance(impl_type, sqltypes.SmallInteger) ) and ( column.default is None or ( isinstance(column.default, schema.Sequence) and column.default.optional )): if isinstance(impl_type, sqltypes.BigInteger): colspec += " BIGSERIAL" elif isinstance(impl_type, sqltypes.SmallInteger): colspec += " SMALLSERIAL" else: colspec += " SERIAL" else: colspec += " " + self.dialect.type_compiler.process(column.type) default = self.get_column_default_string(column) if default is not None: colspec += " DEFAULT " + default if not column.nullable: colspec += " NOT NULL" return colspec
def visit_enum(self, type_): if not type_.native_enum or not self.dialect.supports_native_enum: return super(PGTypeCompiler, self).visit_enum(type_) else: return self.visit_ENUM(type_)
def visit_ENUM(self, type_): return self.dialect.identifier_preparer.format_type(type_)
def get_table_oid(self, table_name, schema=None): """Return the oid from `table_name` and `schema`.""" return self.dialect.get_table_oid(self.bind, table_name, schema, info_cache=self.info_cache)
