SqlAlchemy 2.0 中文文档（四十二）（2）

from sqlalchemy import func
from sqlalchemy.types import UserDefinedType
class Geometry(UserDefinedType):
    def get_col_spec(self):
        return "GEOMETRY"
    def bind_expression(self, bindvalue):
        return func.ST_GeomFromText(bindvalue, type_=self)
    def column_expression(self, col):
        return func.ST_AsText(col, type_=self)

geometry = Table(
    "geometry",
    metadata,
    Column("geom_id", Integer, primary_key=True),
    Column("geom_data", Geometry),
)
print(
    select(geometry).where(
        geometry.c.geom_data == "LINESTRING(189412 252431,189631 259122)"
    )
)

SELECT  geometry.geom_id,  ST_AsText(geometry.geom_data)  AS  geom_data_1
FROM  geometry
WHERE  geometry.geom_data  =  ST_GeomFromText(:geom_data_2)

print(select(geometry.c.geom_data.label("my_data")))

SELECT  ST_AsText(geometry.geom_data)  AS  my_data
FROM  geometry

from sqlalchemy import (
    create_engine,
    String,
    select,
    func,
    MetaData,
    Table,
    Column,
    type_coerce,
    TypeDecorator,
)
from sqlalchemy.dialects.postgresql import BYTEA
class PGPString(TypeDecorator):
    impl = BYTEA
    cache_ok = True
    def __init__(self, passphrase):
        super(PGPString, self).__init__()
        self.passphrase = passphrase
    def bind_expression(self, bindvalue):
        # convert the bind's type from PGPString to
        # String, so that it's passed to psycopg2 as is without
        # a dbapi.Binary wrapper
        bindvalue = type_coerce(bindvalue, String)
        return func.pgp_sym_encrypt(bindvalue, self.passphrase)
    def column_expression(self, col):
        return func.pgp_sym_decrypt(col, self.passphrase)
metadata_obj = MetaData()
message = Table(
    "message",
    metadata_obj,
    Column("username", String(50)),
    Column("message", PGPString("this is my passphrase")),
)
engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test", echo=True)
with engine.begin() as conn:
    metadata_obj.create_all(conn)
    conn.execute(
        message.insert(),
        {"username": "some user", "message": "this is my message"},
    )
    print(
        conn.scalar(select(message.c.message).where(message.c.username == "some user"))
    )

INSERT  INTO  message  (username,  message)
  VALUES  (%(username)s,  pgp_sym_encrypt(%(message)s,  %(pgp_sym_encrypt_1)s))
  -- {'username': 'some user', 'message': 'this is my message',
  --  'pgp_sym_encrypt_1': 'this is my passphrase'}
SELECT  pgp_sym_decrypt(message.message,  %(pgp_sym_decrypt_1)s)  AS  message_1
  FROM  message
  WHERE  message.username  =  %(username_1)s
  -- {'pgp_sym_decrypt_1': 'this is my passphrase', 'username_1': 'some user'}
```## 重新定义和创建新操作符
SQLAlchemy Core 定义了一组固定的表达式运算符，可用于所有列表达式。其中一些操作的效果是重载 Python 的内置运算符；此类运算符的示例包括`ColumnOperators.__eq__()`（`table.c.somecolumn == 'foo'`）、`ColumnOperators.__invert__()`（`~table.c.flag`）和`ColumnOperators.__add__()`（`table.c.x + table.c.y`）。其他运算符作为列表达式的显式方法公开，例如`ColumnOperators.in_()`（`table.c.value.in_(['x', 'y'])`）和`ColumnOperators.like()`（`table.c.value.like('%ed%')`）。
当需要使用尚未直接支持的 SQL 运算符时，最快的方法是在任何 SQL 表达式对象上使用`Operators.op()`方法；此方法接受表示要渲染的 SQL 运算符的字符串，并返回一个 Python 可调用对象，该对象接受任意的右侧表达式：
```py
>>> from sqlalchemy import column
>>> expr = column("x").op(">>")(column("y"))
>>> print(expr)
x  >>  y 

from sqlalchemy import Integer
class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def __add__(self, other):
            return self.op("goofy")(other)

>>> sometable = Table("sometable", metadata, Column("data", MyInt))
>>> print(sometable.c.data + 5)
sometable.data  goofy  :data_1 

from sqlalchemy import Integer
class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def __add__(self, other):
            return func.special_addition(self.expr, other)

from sqlalchemy import Integer, func
class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def log(self, other):
            return func.log(self.expr, other)

>>> print(sometable.c.data.log(5))
log(:log_1,  :log_2) 

class MyInt(Integer):
    class comparator_factory(Integer.Comparator):
        def is_frobnozzled(self, other):
            return self.op("--is_frobnozzled->", is_comparison=True)(other)

from sqlalchemy import Integer
from sqlalchemy.sql.expression import UnaryExpression
from sqlalchemy.sql import operators
class MyInteger(Integer):
    class comparator_factory(Integer.Comparator):
        def factorial(self):
            return UnaryExpression(
                self.expr, modifier=operators.custom_op("!"), type_=MyInteger
            )

>>> from sqlalchemy.sql import column
>>> print(column("x", MyInteger).factorial())
x  ! 

class sqlalchemy.types.UserDefinedType

import sqlalchemy.types as types
class MyType(types.UserDefinedType):
    cache_ok = True
    def __init__(self, precision = 8):
        self.precision = precision
    def get_col_spec(self, **kw):
        return "MYTYPE(%s)" % self.precision
    def bind_processor(self, dialect):
        def process(value):
            return value
        return process
    def result_processor(self, dialect, coltype):
        def process(value):
            return value
        return process

table = Table('foo', metadata_obj,
    Column('id', Integer, primary_key=True),
    Column('data', MyType(16))
    )

attribute cache_ok: bool | None = None

class MyType(TypeDecorator):
    impl = String
    cache_ok = True
    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

class LookupType(UserDefinedType):
  '''a custom type that accepts a dictionary as a parameter.
 this is the non-cacheable version, as "self.lookup" is not
 hashable.
 '''
    def __init__(self, lookup):
        self.lookup = lookup
    def get_col_spec(self, **kw):
        return "VARCHAR(255)"
    def bind_processor(self, dialect):
        # ...  works with "self.lookup" ...

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

>>> # set cache_ok = True
>>> type_.cache_ok = True
>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))
>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

class LookupType(UserDefinedType):
  '''a custom type that accepts a dictionary as a parameter.
 The dictionary is stored both as itself in a private variable,
 and published in a public variable as a sorted tuple of tuples,
 which is hashable and will also return the same value for any
 two equivalent dictionaries.  Note it assumes the keys and
 values of the dictionary are themselves hashable.
 '''
    cache_ok = True
    def __init__(self, lookup):
        self._lookup = lookup
        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple(
            (key, lookup[key]) for key in sorted(lookup)
        )
    def get_col_spec(self, **kw):
        return "VARCHAR(255)"
    def bind_processor(self, dialect):
        # ...  works with "self._lookup" ...

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

method coerce_compared_value(op: OperatorType | None, value: Any) → TypeEngine[Any]

attribute ensure_kwarg: str = 'get_col_spec'

>>> from sqlalchemy import (
...     Table,
...     Column,
...     MetaData,
...     create_engine,
...     PickleType,
...     Integer,
... )
>>> metadata = MetaData()
>>> my_table = Table(
...     "my_table", metadata, Column("id", Integer), Column("data", PickleType)
... )
>>> engine = create_engine("sqlite://", echo="debug")
>>> my_table.create(engine)
INFO  sqlalchemy.engine.base.Engine
CREATE  TABLE  my_table  (
  id  INTEGER,
  data  BLOB
) 

>>> my_table.c.data.type
PickleType()

>>> metadata_two = MetaData()
>>> my_reflected_table = Table("my_table", metadata_two, autoload_with=engine)
INFO  sqlalchemy.engine.base.Engine  PRAGMA  main.table_info("my_table")
INFO  sqlalchemy.engine.base.Engine  ()
DEBUG  sqlalchemy.engine.base.Engine  Col  ('cid',  'name',  'type',  'notnull',  'dflt_value',  'pk')
DEBUG  sqlalchemy.engine.base.Engine  Row  (0,  'id',  'INTEGER',  0,  None,  0)
DEBUG  sqlalchemy.engine.base.Engine  Row  (1,  'data',  'BLOB',  0,  None,  0)
>>>  my_reflected_table.c.data.type
BLOB() 

>>> metadata_three = MetaData()
>>> my_reflected_table = Table(
...     "my_table",
...     metadata_three,
...     Column("data", PickleType),
...     autoload_with=engine,
... )

>>> my_reflected_table.c.data.type
PickleType()

from sqlalchemy import BLOB
from sqlalchemy import event
from sqlalchemy import PickleType
from sqlalchemy import Table
@event.listens_for(Table, "column_reflect")
def _setup_pickletype(inspector, table, column_info):
    if isinstance(column_info["type"], BLOB):
        column_info["type"] = PickleType()

from sqlalchemy.ext.compiler import compiles
from sqlalchemy.types import BINARY
@compiles(BINARY, "sqlite")
def compile_binary_sqlite(type_, compiler, **kw):
    return "BLOB"