def register_enum(
info: EnumInfo,
- enum: Optional[Type[E]] = None,
context: Optional[AdaptContext] = None,
+ enum: Optional[Type[E]] = None,
) -> None:
"""Register the adapters to load and dump a enum type.
:param info: The object with the information about the enum to register.
- :param enum: Python enum type matching to the PostgreSQL one. If `!None`,
- a new enum will be generated and exposed as `EnumInfo.enum`.
:param context: The context where to register the adapters. If `!None`,
register it globally.
+ :param enum: Python enum type matching to the PostgreSQL one. If `!None`,
+ a new enum will be generated and exposed as `EnumInfo.enum`.
"""
if not info:
conn.execute(f"set client_encoding to {encoding}")
name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), enum, conn)
+ register_enum(EnumInfo.fetch(conn, name), conn, enum=enum)
for label in labels:
cur = conn.execute(f"select %{fmt_in}::{name}", [label], binary=fmt_out)
@pytest.mark.parametrize("enum", ascii_cases)
def test_enum_loader_sqlascii(conn, enum, fmt_in, fmt_out):
info = EnumInfo.fetch(conn, enum.__name__.lower())
- register_enum(info, enum, conn)
+ register_enum(info, conn, enum)
conn.execute("set client_encoding to sql_ascii")
for label in info.labels:
conn.execute(f"set client_encoding to {encoding}")
name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), enum, conn)
+ register_enum(EnumInfo.fetch(conn, name), conn, enum)
for item in enum:
cur = conn.execute(f"select %{fmt_in}", [item], binary=fmt_out)
@pytest.mark.parametrize("enum", ascii_cases)
def test_enum_dumper_sqlascii(conn, enum, fmt_in, fmt_out):
info = EnumInfo.fetch(conn, enum.__name__.lower())
- register_enum(info, enum, conn)
+ register_enum(info, conn, enum)
conn.execute("set client_encoding to sql_ascii")
for item in enum:
conn.execute(f"set client_encoding to {encoding}")
name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), enum, conn)
+ register_enum(EnumInfo.fetch(conn, name), conn, enum)
cur = conn.execute(f"select %{fmt_in}::{name}[]", [labels], binary=fmt_out)
assert cur.fetchone()[0] == list(enum)
conn.execute(f"set client_encoding to {encoding}")
name, enum, labels = testenum
- register_enum(EnumInfo.fetch(conn, name), enum, conn)
+ register_enum(EnumInfo.fetch(conn, name), conn, enum)
cur = conn.execute(f"select %{fmt_in}", [list(enum)], binary=fmt_out)
assert cur.fetchone()[0] == list(enum)
await aconn.execute(f"set client_encoding to {encoding}")
name, enum, labels = testenum
- register_enum(await EnumInfo.fetch(aconn, name), enum, aconn)
+ register_enum(await EnumInfo.fetch(aconn, name), aconn, enum)
for label in labels:
cur = await aconn.execute(f"select %{fmt_in}::{name}", [label], binary=fmt_out)