from .types import BIT as BIT
from .types import BYTEA as BYTEA
from .types import CIDR as CIDR
+from .types import CITEXT as CITEXT
from .types import INET as INET
from .types import INTERVAL as INTERVAL
from .types import MACADDR as MACADDR
"real": REAL,
"inet": INET,
"cidr": CIDR,
+ "citext": CITEXT,
"uuid": UUID,
"bit": BIT,
"bit varying": BIT,
return ""
def for_update_clause(self, select, **kw):
-
if select._for_update_arg.read:
if select._for_update_arg.key_share:
tmp = " FOR KEY SHARE"
tmp = " FOR UPDATE"
if select._for_update_arg.of:
-
tables = util.OrderedSet()
for c in select._for_update_arg.of:
tables.update(sql_util.surface_selectables_only(c))
return "SUBSTRING(%s FROM %s)" % (s, start)
def _on_conflict_target(self, clause, **kw):
-
if clause.constraint_target is not None:
# target may be a name of an Index, UniqueConstraint or
# ExcludeConstraint. While there is a separate
return target_text
def visit_on_conflict_do_nothing(self, on_conflict, **kw):
-
target_text = self._on_conflict_target(on_conflict, **kw)
if target_text:
return "ON CONFLICT DO NOTHING"
def visit_on_conflict_do_update(self, on_conflict, **kw):
-
clause = on_conflict
target_text = self._on_conflict_target(on_conflict, **kw)
class PGDDLCompiler(compiler.DDLCompiler):
def get_column_specification(self, column, **kwargs):
-
colspec = self.preparer.format_column(column)
impl_type = column.type.dialect_impl(self.dialect)
if isinstance(impl_type, sqltypes.TypeDecorator):
def visit_CIDR(self, type_, **kw):
return "CIDR"
+ def visit_CITEXT(self, type_, **kw):
+ return "CITEXT"
+
def visit_MACADDR(self, type_, **kw):
return "MACADDR"
return "BYTEA"
def visit_ARRAY(self, type_, **kw):
-
inner = self.process(type_.item_type, **kw)
return re.sub(
r"((?: COLLATE.*)?)$",
class PGIdentifierPreparer(compiler.IdentifierPreparer):
-
reserved_words = RESERVED_WORDS
def _unquote_identifier(self, value):
def get_insert_default(self, column):
if column.primary_key and column is column.table._autoincrement_column:
if column.server_default and column.server_default.has_argument:
-
# pre-execute passive defaults on primary key columns
return self._execute_scalar(
"select %s" % column.server_default.arg, column.type
def get_multi_indexes(
self, connection, schema, filter_names, scope, kind, **kw
):
-
table_oids = self._get_table_oids(
connection, schema, filter_names, scope, kind, **kw
)
from ...testing.provision import drop_all_schema_objects_pre_tables
from ...testing.provision import drop_db
from ...testing.provision import log
+from ...testing.provision import post_configure_engine
from ...testing.provision import prepare_for_drop_tables
from ...testing.provision import set_default_schema_on_connection
from ...testing.provision import temp_table_keyword_args
stmt = stmt.returning(*returning)
return stmt
+
+
+@post_configure_engine.for_db("postgresql")
+def _create_citext_extension(url, engine, follower_ident):
+ with engine.connect() as conn:
+ if conn.dialect.server_version_info >= (13,):
+ conn.execute(text("CREATE EXTENSION IF NOT EXISTS citext"))
+ conn.commit()
from sqlalchemy.dialects.postgresql import array
from sqlalchemy.dialects.postgresql import array_agg
from sqlalchemy.dialects.postgresql import base
+from sqlalchemy.dialects.postgresql import CITEXT
from sqlalchemy.dialects.postgresql import DATEMULTIRANGE
from sqlalchemy.dialects.postgresql import DATERANGE
from sqlalchemy.dialects.postgresql import DOMAIN
__requires__ = ("postgresql_jsonb",)
datatype = JSONB
+
+
+class CITextTest(fixtures.TablesTest):
+ __requires__ = ("citext",)
+ __only_on__ = "postgresql"
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "ci_test_table",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("caseignore_text", CITEXT),
+ )
+
+ def test_citext(self, connection):
+ ci_test_table = self.tables.ci_test_table
+ connection.execute(
+ ci_test_table.insert(),
+ {"caseignore_text": "Hello World"},
+ )
+
+ ret = connection.execute(
+ select(ci_test_table.c.caseignore_text == "hello world")
+ ).scalar()
+
+ assert ret is not None
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
