Download a DataBC WFS layer to postgres - an ogr2ogr wrapper.

     \b
      $ bcdata bc2pg bc-airports --db_url postgresql://postgres:postgres@localhost:5432/postgis

    The default target database can be specified by setting the $DATABASE_URL
    environment variable.
    https://docs.sqlalchemy.org/en/latest/core/engines.html#database-urls

def bc2pg(dataset, db_url, table, schema, query, append, pagesize, sortby, max_workers):
    """Download a DataBC WFS layer to postgres - an ogr2ogr wrapper.

     \b
      $ bcdata bc2pg bc-airports --db_url postgresql://postgres:postgres@localhost:5432/postgis

    The default target database can be specified by setting the $DATABASE_URL
    environment variable.
    https://docs.sqlalchemy.org/en/latest/core/engines.html#database-urls
    """
    src = bcdata.validate_name(dataset)
    src_schema, src_table = [i.lower() for i in src.split(".")]
    if not schema:
        schema = src_schema
    if not table:
        table = src_table

    # create schema if it does not exist
    conn = pgdata.connect(db_url)
    if schema not in conn.schemas:
        click.echo("Schema {} does not exist, creating it".format(schema))
        conn.create_schema(schema)

    # build parameters for each required request
    param_dicts = bcdata.define_request(
        dataset, query=query, sortby=sortby, pagesize=pagesize
    )
    try:
        # run the first request / load
        payload = urlencode(param_dicts[0], doseq=True)
        url = bcdata.WFS_URL + "?" + payload
        db = parse_db_url(db_url)
        db_string = "PG:host={h} user={u} dbname={db} password={pwd}".format(
            h=db["host"], u=db["user"], db=db["database"], pwd=db["password"]
        )

        # create the table
        if not append:
            command = [
                "ogr2ogr",
                "-lco",
                "OVERWRITE=YES",
                "-lco",
                "SCHEMA={}".format(schema),
                "-lco",
                "GEOMETRY_NAME=geom",
                "-f",
                "PostgreSQL",
                db_string,
                "-t_srs",
                "EPSG:3005",
                "-nln",
                table,
                url,
            ]
            click.echo(" ".join(command))
            subprocess.run(command)

        # append to table when append specified or processing many chunks
        if len(param_dicts) > 1 or append:
            # define starting index in list of requests
            if append:
                idx = 0
            else:
                idx = 1
            commands = []
            for chunk, paramdict in enumerate(param_dicts[idx:]):
                payload = urlencode(paramdict, doseq=True)
                url = bcdata.WFS_URL + "?" + payload
                command = [
                    "ogr2ogr",
                    "-update",
                    "-append",
                    "-f",
                    "PostgreSQL",
                    db_string + " active_schema=" + schema,
                    "-t_srs",
                    "EPSG:3005",
                    "-nln",
                    table,
                    url,
                ]
                commands.append(command)

            # https://stackoverflow.com/questions/14533458
            pool = Pool(max_workers)
            with click.progressbar(
                pool.imap(partial(call), commands), length=len(param_dicts)
            ) as bar:
                for returncode in bar:
                    if returncode != 0:
                        click.echo("Command failed: {}".format(returncode))

        click.echo(
            "Load of {} to {} in {} complete".format(src, schema + "." + table, db_url)
        )
    except Exception:
        click.echo("Data load failed")
        raise click.Abort()