def filter(
    url: Union[str, pd.DataFrame],
    args: dict = {},
    meta: dict = {},
    engine: str = None,
    table: str = None,
    ext: str = None,
    id: List[str] = None,
    columns: Dict[str, Union[str, dict]] = None,
    query: str = None,
    queryfile: str = None,
    transform: Callable = None,
    transform_kwargs: dict = {},
    argstype: Dict[str, dict] = {},
    **kwargs: dict,
) -> pd.DataFrame:
    '''Filter data using URL query parameters.

    Examples:
        >>> gramex.data.filter(dataframe, args=handler.args)
        >>> gramex.data.filter('file.csv', args=handler.args)
        >>> gramex.data.filter('mysql://server/db', table='table', args={'user': [user']})

    Parameters:

        url: DataFrame, sqlalchemy URL, directory or file name, http(s) URL
        args: URL query parameters as a dict of lists. Pass handler.args or parse_qs results
        meta: this dict is updated with metadata during the course of filtering
        engine: over-rides the auto-detected engine. Can be 'dataframe', 'file',
            'http', 'https', 'sqlalchemy', 'dir'
        ext: file extension (if url is a file). Defaults to url extension
        columns: database column names to create if required (if url is a database).
            Keys are column names. Values can be SQL types, or dicts with these keys:
                - `type` (str), e.g. `"VARCHAR(10)"`
                - `default` (str/int/float/bool), e.g. `"none@example.org"`
                - `nullable` (bool), e.g. `False`
                - `primary_key` (bool), e.g. `True` -- used only when creating new tables
                - `autoincrement` (bool), e.g. `True` -- used only when creating new tables
        query: optional SQL query to execute (if url is a database),
            `.format`-ed using `args` and supports SQLAlchemy SQL parameters.
            Loads entire result in memory before filtering.
        queryfile: optional SQL query file to execute (if url is a database).
            Same as specifying the `query:` in a file. Overrides `query:`
        transform: optional in-memory transform of source data. Takes
            the result of gramex.cache.open or gramex.cache.query. Must return a
            DataFrame. Applied to both file and SQLAlchemy urls.
        transform_kwargs: optional keyword arguments to be passed to the
            transform function -- apart from data
        argstype: optional dict that specifies `args` type and behavior for `query`.
        **kwargs: Additional parameters are passed to
            [gramex.cache.open][], `sqlalchemy.create_engine` or the plugin's filter

    Returns:
        Filtered DataFrame

    To filter a DataFrame where column x=1 and y=2:

        >>> filtered = gramex.data.filter(dataframe, args={'x': [1], 'y': [2]})

    `args` is always a dict of lists, which is compatible with Gramex's `handler.args`.
    So you can replace the above with:

        >>> filtered = gramex.data.filter(dataframe, args=handle.args)

    To filter ?city=Rome from a CSV/XLS/any file supported by [gramex.cache.open][]:

        >>> gramex.data.filter('path/to/file.csv', rel=True, args={'city': ['Rome']})

    Remaining `kwargs` are passed to [gramex.cache.open][] if `url` is a file. So `rel=True` works

    To filter ?city=Rome from a SQLite, MySQL, PostgreSQL or any SQLAlchemy-supported DB:

        >>> gramex.data.filter('sqlite:///x.db', table='data', args={'city': ['Rome']})

    Remaining `kwargs` are passed to `sqlalchemy.create_engine` if `url` is a SQLAlchemy URL. E.g.

    - `table`: table name (if url is an SQLAlchemy URL), `.format`-ed using `args`.
    - `state`: optional SQL query to check if data has changed.

    TODO: Document how to pass params -- for each database

    If `table` or `query` is passed to an SQLAlchemy url, it is formatted using `args`.
    For example

        >>> data = gramex.data.filter('mysql://server/db', table='{xxx}', args=handler.args)

    ... when passed `?xxx=sales` returns rows from the sales table. Similarly:

        >>> data = gramex.data.filter(
        ...     'mysql://server/db', args=handler.args,
        ...     query='SELECT {col}, COUNT(*) FROM table GROUP BY {col}')

    ... when passsed `?col=City` replaces `{col}` with `City`.

    NOTE: To avoid SQL injection attacks, only keys without spaces are allowed.
    So `?city name=Oslo` **will not** work.

    The URL supports operators filter like this:

    - `?x` selects x is not null
    - `?x!` selects x is null
    - `?x=val` selects x == val
    - `?x!=val` selects x != val
    - `?x>=val` selects x > val
    - `?x>~=val` selects x >= val
    - `?x<=val` selects x < val
    - `?x<~=val` selects x <= val
    - `?x~=val` selects x matches val
    - `?x!~=val` selects x does not match val
    - `?x*=val` selects x matches val case-insensitively
    - `?x!*=val` selects x does not match val case-insensitively

    Multiple filters are combined into an AND clause. Ranges can also be
    specified like this:

    - `?x=a&y=b` selects x = a AND y = b
    - `?x>=100&x<=200` selects x > 100 AND x < 200

    If the same column has multiple values, they are combined like this:

    - `?x=a&x=b` selects x IN (a, b)
    - `?x!=a&x!=b` selects x NOT IN (a, b)
    - `?x~=a&x~=b` selects x ~ a|b
    - `?x>=a&x>=b` selects x > MIN(a, b)
    - `?x<=a&x<=b` selects x < MAX(a, b)

    Arguments are converted to the type of the column before comparing. If this
    fails, it raises a ValueError.

    You can specify the SQL argument type for database queries with `argstype`. This (currently)
    only applies when using `query` or `queryfunction`, not `table`:

    - `argstype: {'x': {type: float}, 'y': {type: bool}}` treats x as a float and y as a bool
    - `argstype: {'x': {type: int, expanding=True}}` treats x as a list of int, suitable for
      use in an `IN` clause, e.g. `SELECT * FROM table WHERE size IN :x`

    These URL query parameters control the output:

    - `?_sort=col` sorts column col in ascending order. `?_sort=-col` sorts
        in descending order.
    - `?_limit=100` limits the result to 100 rows
    - `?_offset=100` starts showing the result from row 100. Default: 0
    - `?_c=x&_c=y` returns only columns `[x, y]`. `?_c=-col` drops col.

    If a column name matches one of the above, you cannot filter by that column.
    Avoid column names beginning with _.

    To get additional information about the filtering, use:

        meta = {}  # Create a variable which will be filled with more info
        filtered = gramex.data.filter(data, meta=meta, **handler.args)

    The `meta` variable is populated with the following keys:

    - `filters`: Applied filters as `[(col, op, val), ...]`
    - `ignored`: Ignored filters as `[(col, vals), ('_sort', col), ('_by', col), ...]`
    - `excluded`: Excluded columns as `[col, ...]`
    - `sort`: Sorted columns as `[(col, True), ...]`. The second parameter is `ascending=`
    - `offset`: Offset as integer. Defaults to 0
    - `limit`: Limit as integer - `None` if limit is not applied
    - `count`: Total number of rows, if available
    - `by`: Group by columns as `[col, ...]`
    - `inserted`: List of (dict of primary values) for each inserted row

    These variables may be useful to show additional information about the
    filtered data.
    '''
    # Auto-detect engine.
    if engine is None:
        engine = get_engine(url)

    # Pass the meta= argument from kwargs (if any)
    meta.update(
        {
            'filters': [],  # Applied filters as [(col, op, val), ...]
            'ignored': [],  # Ignored filters as [(col, vals), ...]
            'sort': [],  # Sorted columns as [(col, asc), ...]
            'offset': 0,  # Offset as integer
            'limit': None,  # Limit as integer - None if not applied
            'by': [],  # Group by columns as [col, ...]
        }
    )
    args = dict(args)  # Do not modify the args -- keep a copy
    controls = _pop_controls(args)
    transform = _transform_fn(transform, transform_kwargs)
    url, ext, query, queryfile, table, kwargs = _replace(
        engine, args, url, ext, query, queryfile, table, **kwargs
    )

    # Use the appropriate filter function based on the engine
    if engine == 'dataframe':
        data = transform(url) if callable(transform) else url
        return _filter_frame(data, meta, controls, args, argstype)
    elif engine == 'dir':
        data = dirstat(url, **args)
        data = transform(data) if callable(transform) else data
        return _filter_frame(data, meta, controls, args, argstype)
    elif engine in {'file', 'http', 'https'}:
        if engine == 'file' and not os.path.exists(url):
            raise OSError(f'url: {url} not found')
        # table= is not a valid option for all gramex.cache.open formats. Use only if specified
        if table is not None:
            kwargs['table'] = table
        # Get the full dataset. Then filter it
        data = gramex.cache.open(url, ext, transform=transform, **kwargs)
        return _filter_frame(data, meta, controls, args, argstype)
    elif engine.startswith('plugin+'):
        plugin = engine.split('+')[1]
        method = plugins[plugin]['filter']
        return method(
            url=url,
            meta=meta,
            controls=controls,
            args=args,
            argstype=argstype,
            id=id,
            table=table,
            columns=columns,
            ext=ext,
            query=query,
            queryfile=queryfile,
            **kwargs,
        )
    elif engine == 'sqlalchemy':
        state = kwargs.pop('state', None)
        engine = alter(url, table, columns, **kwargs)
        if query or queryfile:
            if queryfile:
                query = gramex.cache.open(queryfile, 'text')
            if not state:
                if isinstance(table, str):
                    state = table if ' ' in table else [table]
                elif isinstance(table, (list, tuple)):
                    state = list(table)
                elif table is not None:
                    raise ValueError(f'table: must be string or list of strings, not {table!r}')
            # sa.text() provides backend-neutral :name for bind parameters
            # NOTE: sa.text() caches queries => .bindparams() UPDATES previous bindparams. So:
            #   If query = "SELECT * FROM table WHERE x=:x" and we bind with bindparam('x'),
            #   we can NEVER bind with x as bindparam('x', expanding=True) without restarting.
            #   Therefore, create a new query each time.
            sql = sa.text(f'{query}; -- {time.time()}')
            all_params = {}
            for key, vals in args.items():
                conv, expanding = _argstype(argstype, key, str)
                if expanding:
                    all_params[key] = tuple(conv(val) for val in vals if val)
                    sql = sql.bindparams(sa.bindparam(key, expanding=True))
                elif len(vals) > 0:
                    all_params[key] = conv(vals[0])
            data = gramex.cache.query(sql, engine, state, params=all_params)
            data = transform(data) if callable(transform) else data
            # The query acts as base data. Now filter with additional parameters
            return _filter_frame(data, meta, controls, args, argstype)
        elif table:
            if callable(transform):
                data = gramex.cache.query(table, engine, [table])
                return _filter_frame(transform(data), meta, controls, args, argstype)
            else:
                return _filter_db(engine, table, meta, controls, args, argstype)
        else:
            raise ValueError('No table: or query: specified')
    else:
        raise ValueError(f'engine: {engine} invalid. Can be sqlalchemy|file|dataframe')

def delete(
    url: Union[str, pd.DataFrame],
    args: dict = {},
    meta: dict = {},
    engine: str = None,
    table: str = None,
    ext: str = None,
    id: List[str] = None,
    columns: Dict[str, Union[str, dict]] = None,
    query: str = None,
    queryfile: str = None,
    transform: Callable = None,
    transform_kwargs: dict = {},
    argstype: Dict[str, dict] = {},
    **kwargs: dict,
) -> int:
    '''Deletes data using URL query parameters.

    Examples:
        >>> gramex.data.delete(dataframe, args={'city': ['Oslo']})
        >>> gramex.data.delete('mysql://server/db', table='x', args={'city': ['Oslo']})
        >>> gramex.data.delete(url, args=handler.args)

    It accepts the same parameters as [gramex.data.filter][], and returns the number
    of deleted rows.
    '''
    if engine is None:
        engine = get_engine(url)
    meta.update({'filters': [], 'ignored': []})
    args = dict(args)  # Do not modify the args -- keep a copy
    controls = _pop_controls(args)
    url, table, ext, query, queryfile, kwargs = _replace(
        engine, args, url, table, ext, query, queryfile, **kwargs
    )
    if engine == 'dataframe':
        data_filtered = _filter_frame(url, meta, controls, args, argstype, source='delete', id=id)
        return len(data_filtered)
    elif engine == 'file':
        data = gramex.cache.open(url, ext, transform=transform, **kwargs)
        data_filtered = _filter_frame(data, meta, controls, args, argstype, source='delete', id=id)
        gramex.cache.save(data, url, ext, index=False, **kwargs)
        return len(data_filtered)
    elif engine.startswith('plugin+'):
        plugin = engine.split('+')[1]
        method = plugins[plugin]['delete']
        return method(
            url=url,
            meta=meta,
            controls=controls,
            args=args,
            argstype=argstype,
            id=id,
            table=table,
            columns=columns,
            ext=ext,
            query=query,
            queryfile=queryfile,
            **kwargs,
        )
    elif engine == 'sqlalchemy':
        if table is None:
            raise ValueError('No table: specified')
        engine = alter(url, table, columns, **kwargs)
        return _filter_db(engine, table, meta, controls, args, argstype, source='delete', id=id)
    else:
        raise ValueError(f'engine: {engine} invalid. Can be sqlalchemy|file|dataframe')

def update(
    url: Union[str, pd.DataFrame],
    args: dict = {},
    meta: dict = {},
    engine: str = None,
    table: str = None,
    ext: str = None,
    id: List[str] = None,
    columns: Dict[str, Union[str, dict]] = None,
    query: str = None,
    queryfile: str = None,
    transform: Callable = None,
    transform_kwargs: dict = {},
    argstype: Dict[str, dict] = {},
    **kwargs: dict,
) -> int:
    '''Update data using URL query parameters.

    Examples:
        >>> gramex.data.update(dataframe, args=handler.args, id=['id'])
        >>> gramex.data.update('file.csv', args=handler.args, id=['id'])
        >>> gramex.data.update('mysql://server/db', table='x', args=handler.args, id=['id'])

    `id` is a list of column names defining the primary key.
    Calling this in a handler with `?id=1&x=2` updates x=2 where id=1.

    It accepts the same parameters as [gramex.data.filter][], and returns the number of updated
    rows.
    '''
    if engine is None:
        engine = get_engine(url)
    meta.update({'filters': [], 'ignored': []})
    args = dict(args)  # Do not modify the args -- keep a copy
    controls = _pop_controls(args)
    url, table, ext, query, queryfile, kwargs = _replace(
        engine, args, url, table, ext, query, queryfile, **kwargs
    )
    if engine == 'dataframe':
        data_updated = _filter_frame(url, meta, controls, args, argstype, source='update', id=id)
        return len(data_updated)
    elif engine == 'file':
        data = gramex.cache.open(url, ext, transform=transform, **kwargs)
        data_updated = _filter_frame(data, meta, controls, args, argstype, source='update', id=id)
        gramex.cache.save(data, url, ext, index=False, **kwargs)
        return len(data_updated)
    elif engine.startswith('plugin+'):
        plugin = engine.split('+')[1]
        method = plugins[plugin]['update']
        return method(
            url=url,
            meta=meta,
            controls=controls,
            args=args,
            argstype=argstype,
            id=id,
            table=table,
            columns=columns,
            ext=ext,
            query=query,
            queryfile=queryfile,
            **kwargs,
        )
    elif engine == 'sqlalchemy':
        if table is None:
            raise ValueError('No table: specified')
        engine = alter(url, table, columns, **kwargs)
        return _filter_db(engine, table, meta, controls, args, argstype, source='update', id=id)
    else:
        raise ValueError(f'engine: {engine} invalid. Can be sqlalchemy|file|dataframe')

def insert(
    url: Union[str, pd.DataFrame],
    args: dict = {},
    meta: dict = {},
    engine: str = None,
    table: str = None,
    ext: str = None,
    id: List[str] = None,
    columns: Dict[str, Union[str, dict]] = None,
    query: str = None,
    queryfile: str = None,
    transform: Callable = None,
    transform_kwargs: dict = {},
    argstype: Dict[str, dict] = {},
    **kwargs: dict,
) -> int:
    '''Insert data using URL query parameters.

    Examples:
        >>> gramex.data.insert(dataframe, args=handler.args, id=['id'])
        >>> gramex.data.insert('file.csv', args=handler.args, id=['id'])
        >>> gramex.data.insert('mysql://server/db', table='x', args=handler.args, id=['id'])

    `id` is a list of column names defining the primary key.
    Calling this in a handler with `?id=3&x=2` inserts a new record with id=3 and x=2.

    If the target file / table does not exist, it is created.

    It accepts the same parameters as [gramex.data.filter][], and returns the number of updated
    rows.
    '''
    if engine is None:
        engine = get_engine(url)
    args = dict(args)  # Do not modify the args -- keep a copy
    controls = _pop_controls(args)
    if not args:
        raise ValueError('No args: specified')
    meta.update({'filters': [], 'ignored': [], 'inserted': []})
    # If values do not have equal number of elements, pad them and warn
    rowcount = max(len(val) for val in args.values())
    for key, val in args.items():
        rows = len(val)
        if 0 < rows < rowcount:
            val += [val[-1]] * (rowcount - rows)
            app_log.warning(
                f'data.insert: column {key} has {rows} rows not {rowcount}. '
                f'Extended last value {val[-1]}'
            )
    rows = pd.DataFrame.from_dict(args)
    url, table, ext, query, queryfile, kwargs = _replace(
        engine, args, url, table, ext, query, queryfile, **kwargs
    )
    if engine == 'dataframe':
        rows = _pop_columns(rows, url.columns, meta['ignored'])
        url = url.append(rows, sort=False)
        return len(rows)
    elif engine == 'file':
        try:
            data = gramex.cache.open(url, ext, transform=None, **kwargs)
        except OSError:
            data = rows
        else:
            rows = _pop_columns(rows, data.columns, meta['ignored'])
            data = data.append(rows, sort=False)
        gramex.cache.save(data, url, ext, index=False, **kwargs)
        return len(rows)
    elif engine.startswith('plugin+'):
        plugin = engine.split('+')[1]
        method = plugins[plugin]['insert']
        return method(
            url=url,
            meta=meta,
            controls=controls,
            rows=rows,
            args=args,
            argstype=argstype,
            id=id,
            table=table,
            columns=columns,
            ext=ext,
            query=query,
            queryfile=queryfile,
            **kwargs,
        )
    elif engine == 'sqlalchemy':
        if table is None:
            raise ValueError('No table: specified')
        engine = alter(url, table, columns, **kwargs)
        try:
            cols = get_table(engine, table).columns
        except sa.exc.NoSuchTableError:
            pass
        else:
            rows = _pop_columns(rows, [col.name for col in cols], meta['ignored'])
        if '.' in table:
            kwargs['schema'], table = table.rsplit('.', 1)
        # SQLAlchemy 1.4+ only allows sa.inspect() for all table introspection
        if version.parse(sa.__version__) >= version.parse('1.4'):
            has_table = sa.inspect(engine).has_table(table, schema=kwargs.get('schema'))
        # SQLAlchemy 1.3- does not have .has_table()
        else:
            has_table = engine.dialect.has_table(engine, table, schema=kwargs.get('schema'))
        # If the DB doesn't yet have the table, create it WITH THE PRIMARY KEYS.
        if not has_table and id:
            # Note: pandas does not document get_schema, so it might change.
            engine.execute(pd.io.sql.get_schema(rows, name=table, keys=id, con=engine))

        def insert_method(tbl, conn, keys, data_iter):
            '''Pandas .to_sql() doesn't return inserted row primary keys. Capture it in meta'''
            data = [dict(zip(keys, row)) for row in data_iter]
            # If the ?id= is not provided, Pandas creates a schema based on available columns,
            # without the `id` column. SQLAlchemy won't return inserted_primary_key unless the
            # metadata has a primary key. So, hoping that the table already has a primary key,
            # load table from DB via extend_existing=True.
            sa_table = sa.Table(
                table, tbl.table.metadata, extend_existing=True, autoload_with=engine
            )
            r = conn.execute(sa_table.insert(), data)
            # SQLAlchemy 1.4+ supports inserted_primary_key_rows.
            if hasattr(r, 'inserted_primary_key_rows'):
                ids = r.inserted_primary_key_rows
            # In SQLAlchemy 1.3, only single inserts have an inserted_primary_key.
            elif hasattr(r, 'inserted_primary_key'):
                ids = [r.inserted_primary_key]
            else:
                ids = []
            # Add non-empty IDs as a dict with associated keys.
            # If there are no auto-generated primary keys in the table, no need to return anything.
            id_cols = [col.name for col in sa_table.primary_key]
            for row in ids:
                if row:
                    meta['inserted'].append(dict(zip(id_cols, row)))

        kwargs['method'] = insert_method
        # If user passes ?col= with an empty string, replace with NULL;
        # because, if the column is an INT/FLOAT, type conversion int('') / float('') will fail.
        for col in rows.columns:
            if rows[col].dtype == object:
                rows[col].replace({'': None}, inplace=True)

        # kwargs might contain additonal unexpected values, pass expected arguments explicitly
        pd.io.sql.to_sql(
            rows,
            table,
            engine,
            if_exists='append',
            index=False,
            schema=kwargs.get('schema', None),
            index_label=kwargs.get('index_label', None),
            chunksize=kwargs.get('chunksize', None),
            dtype=kwargs.get('dtype', None),
            method=kwargs.get('method', None),
        )

        return len(rows)
    else:
        raise ValueError(f'engine: {engine} invalid. Can be sqlalchemy|file|dataframe')

def get_engine(url: Union[str, pd.DataFrame]) -> str:
    '''Detect the type of url passed.

    The return value is

    - `'dataframe'` if url is a Pandas DataFrame
    - `'sqlalchemy'` if url is a sqlalchemy compatible URL
    - `'plugin'` if it is `<valid-plugin-name>://...`
    - `protocol` if url is of the form `protocol://...`
    - `'dir'` if it is not a URL but a valid directory
    - `'file'` if it is not a URL but a valid file
    - `None` otherwise
    '''
    if isinstance(url, pd.DataFrame):
        return 'dataframe'
    for plugin_name in plugins:
        if url.startswith(f'{plugin_name}:'):
            return f'plugin+{plugin_name}'
    try:
        url = sa.engine.url.make_url(url)
    except sa.exc.ArgumentError:
        return 'dir' if os.path.isdir(url) else 'file'
    try:
        url.get_driver_name()
        return 'sqlalchemy'
    except sa.exc.NoSuchModuleError:
        return url.drivername

def create_engine(url: str, create: sa.engine.base.Engine = sa.create_engine, **kwargs: dict):
    '''
    Cached version of sqlalchemy.create_engine (or any custom engine).

    Normally, this is not required. But [gramex.data.get_table][] caches the engine
    *and* metadata *and* uses autoload=True. This makes sqlalchemy create a new
    database connection for every engine object, and not dispose it. So we
    re-use the engine objects within this module.
    '''
    if url not in _ENGINE_CACHE:
        _ENGINE_CACHE[url] = create(url, **kwargs)
    return _ENGINE_CACHE[url]

def get_table(engine: sa.engine.base.Engine, table: str, **kwargs: dict) -> sa.Table:
    '''Return the sqlalchemy table from the engine and table name'''
    if engine not in _METADATA_CACHE:
        _METADATA_CACHE[engine] = sa.MetaData(bind=engine)
    metadata = _METADATA_CACHE[engine]
    if '.' in table:
        kwargs['schema'], table = table.rsplit('.', 1)
    return sa.Table(table, metadata, autoload=True, autoload_with=engine, **kwargs)

def download(
    data: Union[pd.DataFrame, List[pd.DataFrame]],
    format: str = 'json',
    template: str = None,
    args: dict = {},
    **kwargs: dict,
) -> bytes:
    '''
    Download a DataFrame or dict of DataFrames in various formats. This is used
    by [gramex.handlers.FormHandler][]. You are **strongly** advised to
    try it before creating your own FunctionHandler.

    Usage as a FunctionHandler:

        def download_as_csv(handler):
            handler.set_header('Content-Type', 'text/csv')
            handler.set_header('Content-Disposition', 'attachment;filename=data.csv')
            return gramex.data.download(dataframe, format='csv')

    It takes the following arguments:

    Parameters:
        data: A DataFrame or a dict of DataFrames
        format: Output format. Can be `csv|json|html|xlsx|template`
        template: Path to template file for `template` format
        args: dictionary of user arguments to subsitute spec
        **kwargs: Additional parameters that are passed to the relevant renderer

    Returns:
        bytes with the download file contents

    When `data` is a DataFrame, this is what different `format=` parameters
    return:

    - `csv` returns a UTF-8-BOM encoded CSV file of the dataframe
    - `xlsx` returns an Excel file with 1 sheet named `data`. kwargs are
        passed to `.to_excel(index=False)`
    - `html` returns a HTML file with a single table. kwargs are passed to
        `.to_html(index=False)`
    - `json` returns a JSON file. kwargs are passed to
        `.to_json(orient='records', force_ascii=True)`.
    - `template` returns a Tornado template rendered file. The template
        receives `data` as `data` and any additional kwargs.
    - `pptx` returns a PPTX generated by pptgen
    - `seaborn` or `sns` returns a Seaborn generated chart
    - `vega` returns JavaScript that renders a Vega chart

    When `data` is a dict of DataFrames, the following additionally happens:

    - `format='csv'` renders all DataFrames one below the other, adding the
        key as heading
    - `format='xlsx'` renders each DataFrame on a sheet whose name is the key
    - `format='html'` renders tables below one another with the key as heading
    - `format='json'` renders as a dict of DataFrame JSONs
    - `format='template'` sends `data` and all `kwargs` as passed to the
        template
    - `format='pptx'` passes `data` as a dict of datasets to pptgen
    - `format='vega'` passes `data` as a dict of datasets to Vega

    When `data` is NEITHER a DataFrame or a dict of DataFrames:

    - `format='json'` renders as JSON if possible
    - `format='template'` renders the template if possible
    - all other formats raise a `ValueError`

    You need to set the MIME types on the handler yourself. Recommended MIME
    types are in gramex.yaml under handler.FormHandler.
    '''
    multiple_datasets = True
    error_no_dataframe = None

    # Check if data is a DataFrame or a dict of DataFrames (multiple_datasets).
    # Ensure that data becomes a dict of DataFrames
    if isinstance(data, dict):
        for key, val in data.items():
            if not isinstance(val, pd.DataFrame):
                error_no_dataframe = f'download(): {key} type is {type(val)}, not a DataFrame'
        if not len(data):
            error_no_dataframe = 'download(): got empty dict. Need a DataFrame'
    elif not isinstance(data, pd.DataFrame):
        error_no_dataframe = f'download(): type is {type(data)}, not a DataFrame'
    else:
        data = {'data': data}
        multiple_datasets = False

    # These formats require a DataFrame or a dict of DataFrames. Other formats (json, template)
    # accept anything.
    if error_no_dataframe and format in {
        'csv',
        'html',
        'xlsx',
        'xls',
        'pptx',
        'ppt',
        'seaborn',
        'sns',
        'vega',
        'vega-lite',
        'vegam',
    }:
        raise ValueError(error_no_dataframe)

    def kw(**conf):
        '''Set provided conf as defaults for kwargs'''
        return merge(kwargs, conf, mode='setdefault')

    if format == 'csv':
        # csv.writer requires BytesIO in PY2 and StringIO in PY3.
        # I can't see an elegant way out of this other than writing code for each.
        out = io.StringIO()
        kw(index=False)
        for index, (key, val) in enumerate(data.items()):
            if index > 0:
                out.write('\n')
            if multiple_datasets:
                out.write(key + '\n')
            val.to_csv(out, **kwargs)
        result = out.getvalue()
        # utf-8-sig encoding returns the result with a UTF-8 BOM. Easier to open in Excel
        return result.encode('utf-8-sig') if result.strip() else result.encode('utf-8')
    elif format == 'template':
        return gramex.cache.open(template, 'template').generate(
            data=data if multiple_datasets else data['data'], **kwargs
        )
    elif format == 'html':
        out = io.StringIO()
        kw(index=False)
        for key, val in data.items():
            if multiple_datasets:
                out.write(f'<h1>{key}</h1>')
            val.to_html(out, **kwargs)
        return out.getvalue().encode('utf-8')
    elif format in {'xlsx', 'xls'}:
        out = io.BytesIO()
        kw(index=False)
        # TODO: Create and use a FrameWriter for formatting
        with pd.ExcelWriter(out, engine='xlsxwriter') as writer:
            for key, val in data.items():
                val.to_excel(writer, sheet_name=key, **kwargs)
        return out.getvalue()
    elif format in {'pptx', 'ppt'}:
        from gramex.pptgen import pptgen

        kw()
        out = io.BytesIO()
        pptgen(target=out, data=data, **kwargs)
        return out.getvalue()
    elif format in {'seaborn', 'sns'}:
        kw = AttrDict()
        defaults = {
            'chart': 'barplot',
            'ext': 'png',
            'data': 'data',
            'dpi': 96,
            'width': 640,
            'height': 480,
        }
        for key, default in defaults.items():
            kw[key] = kwargs.pop(key, default)
        import matplotlib

        matplotlib.use('Agg')  # Before importing seaborn, set a headless backend
        import seaborn as sns

        plot = getattr(sns, kw.chart)(data=data.get(kw.data), **kwargs)
        out = io.BytesIO()
        fig = plot.figure if hasattr(plot, 'figure') else plot.fig
        for k in ['dpi', 'width', 'height']:
            kw[k] = float(kw[k])
        fig.set_size_inches(kw.width / kw.dpi, kw.height / kw.dpi)
        fig.savefig(out, format=kw.ext, dpi=kw.dpi)
        fig.clear()
        return out.getvalue()
    elif format in {'vega', 'vega-lite', 'vegam'}:
        kwargs = kw(orient='records', force_ascii=True)
        spec = kwargs.pop('spec', {})
        kwargs.pop('handler', None)
        out = io.BytesIO()
        # conf = {..., spec: {..., data: __DATA__}}
        if isinstance(spec.get('data'), (dict, list)) or 'fromjson' in spec:
            # support only one dataset
            values = list(data.values())
            out.write(values[0].to_json(**kwargs).encode('utf-8'))
            out = out.getvalue()
        else:
            spec['data'] = '__DATA__'
            for index, (key, val) in enumerate(data.items()):
                out.write(b',{"name":' if index > 0 else b'{"name":')
                out.write(json_encode(key).encode('utf-8'))
                out.write(b',"values":')
                out.write(val.to_json(**kwargs).encode('utf-8'))
                out.write(b'}')
            out = out.getvalue()
            if format == 'vega':
                out = b'[' + out + b']'
        kwargs['spec'], _ = _replace('', args, spec)
        conf = json.dumps(kwargs, ensure_ascii=True, separators=(',', ':'), indent=None)
        conf = conf.encode('utf-8').replace(b'"__DATA__"', out)
        script = gramex.cache.open(_VEGA_SCRIPT, 'bin')
        return script.replace(b'/*{conf}*/', conf)
    # If it's none of these formats, default to JSON.
    # If there are no DataFrames, handle arbitrary JSON
    elif error_no_dataframe:
        from gramex.config import CustomJSONEncoder

        kw(cls=CustomJSONEncoder)
        return json.dumps(data, **kwargs)
    # If there ARE DataFrames, render each
    else:
        out = io.BytesIO()
        kwargs = kw(orient='records', force_ascii=True)
        if multiple_datasets:
            out.write(b'{')
            for index, (key, val) in enumerate(data.items()):
                if index > 0:
                    out.write(b',')
                out.write(json_encode(key).encode('utf-8'))
                out.write(b':')
                out.write(val.to_json(**kwargs).encode('utf-8'))
            out.write(b'}')
        else:
            out.write(data['data'].to_json(**kwargs).encode('utf-8'))
        return out.getvalue()

def dirstat(url: str, timeout: int = 10, **kwargs: dict) -> pd.DataFrame:
    '''Return a DataFrame with the list of all files & directories under the url.

    Parameters:
        url: path to a directory, or a URL like `dir:///c:/path/`, `dir:////root/dir/`
        timeout: max seconds to wait. `None` to wait forever

    Raises:
        OSError: if url points to a missing location or is not a directory.

    Returns:
        DataFrame with 1 row per file/directory in the URL

    The result has these columns.

    - `type`: extension with a `.` prefix -- or `dir`
    - `dir`: directory path to the file relative to the URL
    - `name`: file name (including extension)
    - `path`: full path to file or dir. This equals url / dir / name
    - `size`: file size
    - `mtime`: last modified time in seconds since epoch
    - `level`: path depth (i.e. the number of paths in dir)
    '''
    try:
        url = sa.engine.url.make_url(url)
        target = url.database
    except sa.exc.ArgumentError:
        target = url
    if not os.path.isdir(target):
        raise OSError(f'dirstat: {target} is not a directory')
    target = os.path.normpath(target)
    result = []
    start_time = time.time()
    for dirpath, dirnames, filenames in os.walk(target):
        if timeout and time.time() - start_time > timeout:
            app_log.debug(f'dirstat: {url} timeout ({timeout:.1f}s)')
            break
        for name in dirnames:
            path = os.path.join(dirpath, name)
            stat = os.stat(path)
            dirname = dirpath.replace(target, '').replace(os.sep, '/') + '/'
            result.append(
                {
                    'path': path,
                    'dir': dirname,
                    'name': name,
                    'type': 'dir',
                    'size': stat.st_size,
                    'mtime': stat.st_mtime,
                    'level': dirname.count('/'),
                }
            )
        for name in filenames:
            path = os.path.join(dirpath, name)
            stat = os.stat(path)
            dirname = dirpath.replace(target, '').replace(os.sep, '/') + '/'
            result.append(
                {
                    'path': path,
                    'dir': dirname,
                    'name': name,
                    'type': os.path.splitext(name)[-1],
                    'size': stat.st_size,
                    'mtime': stat.st_mtime,
                    'level': dirname.count('/'),
                }
            )
    return pd.DataFrame(result)

def filtercols(
    url: Union[str, pd.DataFrame],
    args: dict = {},
    meta: dict = {},
    engine: str = None,
    ext: str = None,
    query: str = None,
    queryfile: str = None,
    transform: Callable = None,
    transform_kwargs: dict = {},
    separator: str = ',',
    in_memory: bool = False,
    **kwargs: dict,
) -> pd.DataFrame:
    '''Filter data and extract unique values of each column using URL query parameters.

    Examples:
        >>> gramex.data.filtercols(dataframe, args=handler.args)
        >>> gramex.data.filtercols('file.csv', args=handler.args)
        >>> gramex.data.filtercols('mysql://server/db', table='table', args=handler.args)

    Parameters:

        url: Pandas DataFrame, sqlalchemy URL, directory or file name,
            `.format`-ed using `args`.
        args: URL query parameters as a dict of lists. Pass handler.args or parse_qs results
        meta: this dict is updated with metadata during the course of filtering
        engine: over-rides the auto-detected engine. Can be 'dataframe', 'file',
            'http', 'https', 'sqlalchemy', 'dir'
        ext: file extension (if url is a file). Defaults to url extension
        query: optional SQL query to execute (if url is a database),
            `.format`-ed using `args` and supports SQLAlchemy SQL parameters.
            Loads entire result in memory before filtering.
        queryfile: optional SQL query file to execute (if url is a database).
            Same as specifying the `query:` in a file. Overrides `query:`
        transform (function): optional in-memory transform of source data. Takes
            the result of gramex.cache.open or gramex.cache.query. Must return a
            DataFrame. Applied to both file and SQLAlchemy urls.
        transform_kwargs: optional keyword arguments to be passed to the
            transform function -- apart from data
        separator: string that separates columns in a hierarchy. Defaults to `,`.
            For example, `?_c=a,b` treats columns `a` and `b` as a tuple / hierarchy and
            filters them *together*.
        in_memory: fetch all unique values and compute filters in-memory. Faster,
            but takes more memory.
        **kwargs: Additional parameters are passed to
            [gramex.cache.open][] or `sqlalchemy.create_engine`

    Returns:
        filtered DataFrame

    Remaining kwargs are passed to [gramex.cache.open][] if `url` is a file, or
    `sqlalchemy.create_engine` if `url` is a SQLAlchemy URL.

    If this is used in a handler as

        >>> filtered = gramex.data.filtercols(dataframe, args=handler.args)

    ... then calling the handler with `?_c=state&_c=district` returns all unique values
    in columns of `dataframe` where columns are state and district.

    Column filter is supported like this:

    - `?_c=y&x` returns df with unique values of y where x is not null
    - `?_c=y&x=val` returns df with unique values of y where x == val
    - `?_c=y&y=val` returns df with unique values of y, ignores filter y == val
    - `?_c=y&x>=val` returns df with unique values of y where x > val
    - `?_c=x&_c=y&x=val` returns df with unique values of x ignoring filter x == val
        and returns unique values of y where x == val

    Arguments are converted to the type of the column before comparing. If this
    fails, it raises a ValueError.

    These URL query parameters control the output:

    - `?_sort=col` sorts column col in ascending order. `?_sort=-col` sorts
        in descending order.
    - `?_limit=100` limits the result to 100 rows
    - `?_offset=100` starts showing the result from row 100. Default: 0
    - `?_c=x&_c=y` returns only columns `[x, y]`. `?_c=-col` drops col.

    If a column name matches one of the above, you cannot filter by that column.
    Avoid column names beginning with _.

    You can handle hierarchies by passing `?_c=Country,City` with a comma. This returns all unique
    *combinations* of `Country` and `City`.

    To get the min/max or a column, use aggregations, e.g. `?_c=age|min&_c=age|max`.
    You can use `?_c=age|range` as a shortcut that returns min and max of a column.

    To get additional information about the filtering, use:

        meta = {}      # Create a variable which will be filled with more info
        filtered = gramex.data.filter(data, meta=meta, **handler.args)

    The `meta` variable is populated with the following keys:

    - `filters`: Applied filters as `[(col, op, val), ...]`
    - `ignored`: Ignored filters as `[(col, vals), ('_sort', cols), ...]`
    - `excluded`: Excluded columns as `[col, ...]`
    - `sort`: Sorted columns as `[(col, True), ...]`. The second parameter is `ascending=`
    - `offset`: Offset as integer. Defaults to 0
    - `limit`: Limit as integer - `100` if limit is not applied
    - `count`: Total number of rows, if available

    These variables may be useful to show additional information about the
    filtered data.
    '''
    # Auto-detect engine.
    if engine is None:
        engine = get_engine(url)
    result = {}
    limit = args.get('_limit', [100])
    try:
        limit = min(int(v) for v in limit)
    except ValueError:
        raise ValueError(f'_limit not integer: {limit!r}')
    if in_memory:
        # Fetch the superset data, i.e. all filter columns
        in_memory_args = {'_c': [], '_by': set()}
        for col in args.get('_c', []):
            name, agg = col.rsplit(_agg_sep, 1) if _agg_sep in col else (col, None)
            for c in name.split(separator):
                in_memory_args['_by'].add(c)
        # Apply all filters while fetching, if skip the filter columns.
        # We'll apply THOSE filters independently
        for key, vals in args.items():
            col = key
            for op in operators:
                if col.endswith(op):
                    col = col[: -len(op)]
                    break
            if _agg_sep in col:
                col = col.rsplit(_agg_sep, 1)[0]
            if col not in in_memory_args['_by']:
                in_memory_args[key] = vals
        url = filter(url, args=in_memory_args, **kwargs)

    # Get unique values for each column
    for col in args.get('_c', []):
        # If ?_c=sales|RANGE, get the range
        name, agg = col.rsplit(_agg_sep, 1) if _agg_sep in col else (col, None)
        # If ?_c=a,b then treat columns a and b as a pair
        cols = name.split(separator)
        # col_args takes _sort, _c and all filters from args
        col_args = {}
        for key, value in args.items():
            # Apply only _sort as a control. Ignore _by, _limit, _offset, etc.
            if key in ['_sort']:
                col_args[key] = value
            # Apply filters. But ignore filters on the columns we're currently processing
            if not key.startswith('_') and key not in cols:
                col_args[key] = value
        if agg:
            # Convert ?_c=col|RANGE to ?_c=col|MIN&_c=col|MAX
            aggs = ['min', 'max'] if agg.lower() == 'range' else [agg]
            # Group by all values, just return the aggregations
            col_args['_by'] = ['']
            col_args['_c'] = [f'{c}{_agg_sep}{a}' for c in cols for a in aggs]
        else:
            col_args['_by'] = cols
            col_args['_c'] = []
            col_args['_limit'] = [limit]
        result[col] = gramex.data.filter(url, args=col_args, **kwargs)
    return result

def alter(
    url: str, table: str, columns: Dict[str, Union[str, dict]] = None, **kwargs: dict
) -> sa.engine.base.Engine:
    '''Create or alter a table with columns specified.

    Examples:
        >>> gramex.data.alter(url, table, columns={
        ...     'id': {'type': 'int', 'primary_key': True, 'autoincrement': True},
        ...     'when': {'type': 'timestamp', 'default': {'function': 'func.now()'}},
        ...     'email': {'nullable': True, 'default': 'none'},
        ...     'age': {'type': 'float', 'nullable': False, 'default': 18},
        ... })

    Parameters:
        url: sqlalchemy URL
        table: table name
        columns: column names, with values as SQL types or type objects
        **kwargs: passed to `sqlalchemy.create_engine()`.

    Returns:
        SQLAlchemy engine

    If the table exists, new columns (if any) are added. Existing columns are **NOT changed**.

    If the table does not exist, the table is created with the specified columns.

    If there are no changes, the function returns quickly (5 ms), since metadata is cached.

    `columns` can be a dict with values as SQL types (e.g. `"INTEGER"` or `"VARCHAR(10)"`):

        >>> gramex.data.alter(url, table, columns={'id': 'INTEGER', 'name': 'VARCHAR(10)'})

    ... or a dict like `{column_name: type, column_name: type, ...}`:

        >>> gramex.data.alter(url, table, columns={
        ...     'id': {'type': 'int', 'primary_key': True, 'autoincrement': True},
        ...     'when': {'type': 'timestamp', 'default': {'function': 'func.now()'}},
        ...     'email': {'nullable': True, 'default': 'none'},
        ...     'age': {'type': 'float', 'nullable': False, 'default': 18},
        ... })

    If the `columns` values are a dict, these keys are allowed:

    - `type` (str): SQL type, e.g. `"VARCHAR(10)"`
    - `default` (str/int/float/bool/function/dict):
        - A scalar like `"none@example.org"` for fixed default values
        - A SQLAlchemy function like `sqlalchemy.func.now()`
        - A dict like `{function: func.now()}` containing a SQLAlchemy functions
    - `nullable` (bool): whether column can have null values, e.g. `False`
    - `primary_key` (bool): whether column is a primary key, e.g. `True`
    - `autoincrement` (bool): whether column automatically increments, e.g. `True`

    `primary_key` and `autoincrement` are used **only** when creating new tables. They do not
    change existing primary keys or autoincrements. This is because
    [SQLite disallows PRIMARY KEY with ALTER](https://stackoverflow.com/a/1120030/100904)
    and AUTO_INCREMENT doesn't work without PRIMARY KEY in MySQL.
    '''
    engine = create_engine(url, **kwargs)
    if columns is None:
        return engine
    # alter is not required for schema-less databases. For now, hard-code engine names
    scheme = urlparse(url).scheme
    if scheme in {'mongodb', 'elasticsearch', 'influxdb'}:
        app_log.info(f'alter() not required for schema-less DB {engine.driver}')
        return engine
    try:
        db_table = get_table(engine, table, extend_existing=True)
    except sa.exc.NoSuchTableError:
        # If the table's not in the DB, create it
        cols = []
        for name, row in columns.items():
            row = dict({'type': row} if isinstance(row, str) else row, name=name)
            col_type = row.get('type', 'text')
            if isinstance(col_type, str):
                # Use eval() to handle direct types like INTEGER *and* expressions like VARCHAR(3)
                # eval() is safe here since `col_type` is written by app developer
                # B307:eval is safe here since `col_type` is written by app developer
                row['type'] = eval(col_type.upper(), vars(sa.types))  # nosec B307
            row['type_'] = row.pop('type')
            if 'default' in row:
                from inspect import isclass

                default = row.pop('default')
                # default: can be a string like `'sa.func.now()'` or `'func.now()'`
                if isinstance(default, dict):
                    libs = {'sa': sa, 'sqlalchemy': sa, 'func': sa.func}
                    row['server_default'] = build_transform(
                        default, vars={key: None for key in libs}, iter=False
                    )(**libs)
                # default can be an SQLAlchemy function, e.g. sa.func.now()
                elif isclass(default) and issubclass(default, sa.func.Function):
                    row['server_default'] = default
                # default can also be a static value, e.g. `0`
                else:
                    row['server_default'] = str(default)
            cols.append(sa.Column(**row))
        sa.Table(table, _METADATA_CACHE[engine], *cols, extend_existing=True).create(engine)
    else:
        quote = engine.dialect.identifier_preparer.quote_identifier
        # If the table's already in the DB, add new columns. We can't change column types
        with engine.connect() as conn, conn.begin():
            for name, row in columns.items():
                if name in db_table.columns:
                    continue
                row = {'type': row} if isinstance(row, str) else row
                col_type = row.get('type', 'text')
                constraints = []
                if 'nullable' in row:
                    constraints.append('' if row['nullable'] else 'NOT NULL')
                if 'default' in row:
                    # repr() converts int, float properly,
                    #   str into 'str' with single quotes (which is the MySQL standard)
                    #   TODO: datetime and other types will fail
                    if isinstance(row['default'], dict) or callable(row['default']):
                        app_log.warning(
                            f'alter(): col {name} cannot change default on existing table'
                        )
                    else:
                        constraints += ['DEFAULT', repr(row['default'])]
                # This syntax works on DB2, MySQL, Oracle, PostgreSQL, SQLite
                conn.execute(
                    f'ALTER TABLE {quote(table)} '
                    f'ADD COLUMN {quote(name)} {col_type} {" ".join(constraints)}'
                )
        # Refresh table metadata after altering
        get_table(engine, table, extend_existing=True)
    return engine