def _load_reflection_cache(
        self,
        connection: asyncpg.Connection,
    ) -> FrozenSet[str]:
        data = await connection.fetch('''
            SELECT
                eql_hash,
                argnames
            FROM
                ROWS FROM(edgedb._get_cached_reflection())
                    AS t(eql_hash text, argnames text[])
        ''')

        return immutables.Map({
            r['eql_hash']: tuple(r['argnames']) for r in data
        }) 

def ensure_initialized(self, con: asyncpg.Connection) -> None:
        if self._std_schema is None:
            self._std_schema = await load_cached_schema(con, 'stdschema')

        if self._refl_schema is None:
            self._refl_schema = await load_cached_schema(con, 'reflschema')

        if self._schema_class_layout is None:
            self._schema_class_layout = await load_schema_class_layout(con)

        if self._intro_query is None:
            self._intro_query = await load_schema_intro_query(con)

        if self._config_spec is None:
            self._config_spec = config.load_spec_from_schema(
                self._std_schema)
            config.set_settings(self._config_spec) 

def _instrument(self, **kwargs):
        tracer_provider = kwargs.get(
            "tracer_provider", trace.get_tracer_provider()
        )
        setattr(
            asyncpg,
            _APPLIED,
            tracer_provider.get_tracer("asyncpg", __version__),
        )
        for method in [
            "Connection.execute",
            "Connection.executemany",
            "Connection.fetch",
            "Connection.fetchval",
            "Connection.fetchrow",
        ]:
            wrapt.wrap_function_wrapper(
                "asyncpg.connection", method, _do_execute
            ) 

def update_world(self, conn: PoolConn, world: str, update_self=True) -> bool:
        """Updates the world of the character on the database.

        :param conn: Connection to the database.
        :param world: The new world to set.
        :param update_self: Whether to also update the object or not.
        :return: Whether the world was updated in the database or not.
        """
        result = await self.update_field_by_id(conn, self.id, "world", world)
        if result and update_self:
            self.world = world
        return result is not None

    # endregion

    # region Class methods 

def get_latest(cls, conn: PoolConn, *, minimum_level=0, user_id=0, worlds: Union[List[str], str] = None):
        """Gets an asynchronous generator of the character's level ups.

        :param conn: Connection to the database.
        :param minimum_level: The minimum level to show.
        :param user_id: The id of an user to only show level ups of characters they own.
        :param worlds: A list of worlds to only show level ups of characters in that world.
        :return: An asynchronous generator containing the levels.
        """
        if isinstance(worlds, str):
            worlds = [worlds]
        if not worlds:
            worlds = []
        async with conn.transaction():
            async for row in conn.cursor("""
                    SELECT l.*, (json_agg(c)->>0)::jsonb as char FROM character_levelup l
                    LEFT JOIN "character" c ON c.id = l.character_id
                    WHERE ($1::bigint = 0 OR c.user_id = $1) AND (cardinality($2::text[]) = 0 OR c.world = any($2))
                    AND l.level >= $3
                    GROUP BY l.id
                    ORDER BY date DESC""", user_id, worlds, minimum_level):
                yield cls(**row) 

def import_roles(conn: asyncpg.Connection, c: sqlite3.Cursor):
    log.info("Importing roles...")
    auto_roles = []
    joinable_roles = []
    log.debug("Gathering auto roles from sqlite...")
    c.execute("SELECT server_id, role_id, guild FROM auto_roles")
    rows = c.fetchall()
    for server_id, role_id, guild in rows:
        auto_roles.append((server_id, role_id, guild))
    log.debug(f"Collected {len(auto_roles):,} records from old database.")
    log.info("Copying records to auto roles table")
    res = await conn.copy_records_to_table("role_auto", records=auto_roles, columns=["server_id", "role_id", "rule"])
    log.info(f"Copied {get_affected_count(res):,} records successfully.")

    log.debug("Gathering joinable roles from sqlite...")
    c.execute("SELECT server_id, role_id FROM joinable_roles")
    rows = c.fetchall()
    for server_id, role_id in rows:
        joinable_roles.append((server_id, role_id))
    log.debug(f"Collected {len(joinable_roles):,} records from old database.")
    log.info("Copying records to joinable roles table")
    res = await conn.copy_records_to_table("role_joinable", records=joinable_roles, columns=["server_id", "role_id"])
    log.info(f"Copied {get_affected_count(res):,} records successfully.")
    log.info("Finished importing roles.") 

def get_by_name(cls, conn: PoolConn, channel_id: int, name: str, is_guild: bool) -> \
            Optional['WatchlistEntry']:
        """Gets an entry by its name.

        :param conn: Connection to the database.
        :param channel_id: The id of the channel.
        :param name: Name of the entry.
        :param is_guild: Whether the entry is a guild or a character.
        :return: The entry if found.
        """
        row = await conn.fetchrow("SELECT * FROM watchlist_entry "
                                  "WHERE channel_id = $1 AND lower(name) = $2 AND is_guild = $3",
                                  channel_id, name.lower().strip(), is_guild)
        if row is None:
            return None
        return cls(**row) 

def insert(cls, conn: PoolConn, channel_id: int, name: str, is_guild: bool, user_id: int, reason=None)\
            -> Optional['WatchlistEntry']:
        """Inserts a watchlist entry into the database.

        :param conn: Connection to the database.
        :param channel_id: The id of the watchlist's channel.
        :param name: Name of the entry.
        :param is_guild:  Whether the entry is a guild or a character.
        :param user_id: The id of the user that added the entry.
        :param reason: The reason for the entry.
        :return: The inserted entry.
        """
        row = await conn.fetchrow("INSERT INTO watchlist_entry(channel_id, name, is_guild, reason, user_id) "
                                  "VALUES($1, $2, $3, $4, $5) RETURNING *", channel_id, name, is_guild, reason, user_id)
        if row is None:
            return None
        return cls(**row)

# endregion 

def add_listener(self, channel, callback):
        """Add a listener for Postgres notifications.

        :param str channel: Channel to listen on.

        :param callable callback:
            A callable receiving the following arguments:
            **connection**: a Connection the callback is registered with;
            **pid**: PID of the Postgres server that sent the notification;
            **channel**: name of the channel the notification was sent to;
            **payload**: the payload.
        """
        self._check_open()
        if channel not in self._listeners:
            await self.fetch('LISTEN {}'.format(utils._quote_ident(channel)))
            self._listeners[channel] = set()
        self._listeners[channel].add(callback) 

def add_log_listener(self, callback):
        """Add a listener for Postgres log messages.

        It will be called when asyncronous NoticeResponse is received
        from the connection.  Possible message types are: WARNING, NOTICE,
        DEBUG, INFO, or LOG.

        :param callable callback:
            A callable receiving the following arguments:
            **connection**: a Connection the callback is registered with;
            **message**: the `exceptions.PostgresLogMessage` message.

        .. versionadded:: 0.12.0
        """
        if self.is_closed():
            raise exceptions.InterfaceError('connection is closed')
        self._log_listeners.add(callback) 

def introspect(
        self,
        connection: asyncpg.Connection,
    ) -> s_schema.Schema:
        data = await connection.fetch(self._intro_query)
        return s_refl.parse_into(
            schema=self._std_schema,
            data=[r[0] for r in data],
            schema_class_layout=self._schema_class_layout,
        ) 

def __init__(self, conn: Connection):
        super().__init__(conn)
        self._users_repo = UsersRepository(conn) 

def __init__(self, conn: Connection) -> None:
        super().__init__(conn)
        self._profiles_repo = ProfilesRepository(conn) 

def __init__(self, conn: Connection) -> None:
        super().__init__(conn)
        self._profiles_repo = ProfilesRepository(conn)
        self._tags_repo = TagsRepository(conn) 

def init_table(connection: asyncpg.Connection):
    query = """
    CREATE TABLE IF NOT EXISTS STORY (
        id BIGINT PRIMARY KEY,
        content bytea NOT NULL
    );
    """
    await connection.execute(query)
    query = "DELETE FROM STORY;"
    await connection.execute(query) 

def noop_upgrade(_: asyncpg.Connection) -> None:
    pass 

def __aenter__(self) -> asyncpg.Connection: ... 

def __await__(self) -> Awaitable[asyncpg.Connection]: ... 

def _uninstrument(self, **__):
        delattr(asyncpg, _APPLIED)
        for method in [
            "execute",
            "executemany",
            "fetch",
            "fetchval",
            "fetchrow",
        ]:
            unwrap(asyncpg.Connection, method) 

def test_duplicated_instrumentation(self):
        AsyncPGInstrumentor().instrument()
        AsyncPGInstrumentor().instrument()
        AsyncPGInstrumentor().instrument()
        AsyncPGInstrumentor().uninstrument()
        for method_name in ["execute", "fetch"]:
            method = getattr(Connection, method_name, None)
            self.assertFalse(
                hasattr(method, "_opentelemetry_ext_asyncpg_applied")
            ) 

def test_duplicated_uninstrumentation(self):
        AsyncPGInstrumentor().instrument()
        AsyncPGInstrumentor().uninstrument()
        AsyncPGInstrumentor().uninstrument()
        AsyncPGInstrumentor().uninstrument()
        for method_name in ["execute", "fetch"]:
            method = getattr(Connection, method_name, None)
            self.assertFalse(
                hasattr(method, "_opentelemetry_ext_asyncpg_applied")
            ) 

def set_prefixes(pool: PoolConn, guild_id: int, prefixes: List[str]):
    """Sets the new server prefixes.

    :param pool: An asyncpg Pool or Connection.
    :param guild_id: The id of the guild.
    :param prefixes: The list of prefixes to set.
    """
    await pool.execute("""INSERT INTO server_prefixes(server_id, prefixes) VALUES($1, $2)
                          ON CONFLICT(server_id) DO UPDATE SET prefixes = EXCLUDED.prefixes""", guild_id, prefixes) 

def get_server_property(pool: PoolConn, guild_id: int, key: str, default=None) -> Any:
    """Gets the value of a server's property.

    :param pool: An asyncpg Pool or Connection.
    :param guild_id: The id of the guild.
    :param key: The property's key.
    :param default: The value to return if the key has no value.
    :return: The value of the key or the default value if specified.
    """
    value = await pool.fetchval("SELECT value FROM server_property WHERE server_id = $1 AND key = $2", guild_id, key)
    return value if value is not None else default 

def set_server_property(pool: PoolConn, guild_id: int, key: str, value: Any):
    """Sets a server's property.

    :param pool: An asyncpg Pool or Connection.
    :param guild_id: The id of the guild.
    :param key: The property's key.
    :param value: The value to set to the property.
    """
    await pool.execute("""INSERT INTO server_property(server_id, key, value) VALUES($1, $2, $3::jsonb)
                          ON CONFLICT(server_id, key) DO UPDATE SET value = EXCLUDED.value""",
                       guild_id, key, value) 

def get_global_property(pool: PoolConn, key: str, default=None) -> Any:
    """Gets the value of a global property.

    :param pool: An asyncpg Pool or Connection.
    :param key: The property's key
    :param default: The value to return if the property is undefined.
    :return: The value of the key or the default value if specified.
    """
    value = await pool.fetchval("SELECT value FROM global_property WHERE key = $1", key)
    return value if value is not None else default 

def set_global_property(pool: PoolConn, key: str, value: Any):
    """Sets the value of a global property.

    :param pool: An asyncpg Pool or Connection.
    :param key: The property's key
    :param value: The new value the key will have.
    """
    await pool.execute("""INSERT INTO global_property(key, value) VALUES($1, $2::jsonb)
                          ON CONFLICT(key) DO UPDATE SET value = EXCLUDED.value""", key, value) 

def get_level_ups(self, conn: PoolConn):
        """Gets an asynchronous generator of the character's level ups.

        Note that the yielded deaths won't have the char attribute set.

        :param conn: Connection to the database.
        :return: An asynchronous generator containing the entries.
        """
        async for level_up in DbLevelUp.get_from_character(conn, self.id):
            yield level_up 

def get_timeline(self, conn: PoolConn):
        """Gets an asynchronous generator of character's recent deaths and level ups.

        :param conn: Connection to the database.
        :return: An asynchronous generator containing the entries.
        """
        async with conn.transaction():
            async for row in conn.cursor(f"""
                    (
                        SELECT d.*, json_agg(k)::jsonb as killers, 'd' AS type
                        FROM character_death d
                        LEFT JOIN {DbKiller.table} k ON k.death_id = d.id
                        WHERE d.character_id = $1
                        GROUP BY d.id
                    )
                    UNION
                    (
                        SELECT l.*, NULL, 'l' AS type
                        FROM character_levelup l
                        WHERE l.character_id = $1
                        GROUP BY l.id
                    )
                    ORDER by date DESC
                    """, self.id):
                if row["type"] == "l":
                    yield DbLevelUp(**row)
                else:
                    yield DbDeath(**row) 

def update_guild(self, conn: PoolConn, guild: str, update_self=True) -> bool:
        """Updates the guild of the character on the database.

        :param conn: Connection to the database.
        :param guild: The new guild to set.
        :param update_self: Whether to also update the object or not.
        :return: Whether the guild was updated in the database or not.
        """
        result = await self.update_field_by_id(conn, self.id, "guild", guild)
        if result and update_self:
            self.guild = guild
        return result is not None 

def update_level(self, conn: PoolConn, level: int, update_self=True) -> bool:
        """Updates the level of the character on the database.

        :param conn: Connection to the database.
        :param level: The new level to set.
        :param update_self: Whether to also update the object or not.
        :return: Whether the level was updated in the database or not.
        """
        result = await self.update_field_by_id(conn, self.id, "level", level)
        if result and update_self:
            self.level = level
        return result is not None