Method sqlite3

functional/streams.py:247–271

Reads input by querying from a sqlite database. >>> seq.sqlite3('examples/users.db', 'select id, name from users where id = 1;').first() [(1, 'Tom')] :param conn: path or sqlite connection, cursor :param sql: SQL query string :param parameters: Para

(self, conn, sql, parameters=None, *args, **kwargs)

    # pylint: disable=keyword-arg-before-vararg
    def sqlite3(self, conn, sql, parameters=None, *args, **kwargs):
        """
        Reads input by querying from a sqlite database.

        >>> seq.sqlite3('examples/users.db', 'select id, name from users where id = 1;').first()
        [(1, 'Tom')]

        :param conn: path or sqlite connection, cursor
        :param sql: SQL query string
        :param parameters: Parameters for sql query
        :return: Sequence wrapping SQL cursor
        """

        if parameters is None:
            parameters = ()

        if isinstance(conn, (sqlite3api.Connection, sqlite3api.Cursor)):
            return self(conn.execute(sql, parameters))
        elif isinstance(conn, str):
            with sqlite3api.connect(conn, *args, **kwargs) as input_conn:
                return self(input_conn.execute(sql, parameters))
        else:
            raise ValueError(
                "conn must be a must be a file path or sqlite3 Connection/Cursor"
            )


class ParallelStream(Stream):