Stream git history policy changes to destination.


    Default stream destination is a summary of the policy changes to stdout, one
    per line. Also supported for stdout streaming is `jsonline`.

    AWS Kinesis and SQS destinations are specified by providing the ARN.

    Database destinations are supported by providing a sqlalchemy DSN. Note
    SQLAlchemy and db drivers must be installed separately as they an optional
    dependency.

    When using database destinations, streaming defaults to incremental.

def stream(repo_uri, stream_uri, verbose, assume, sort, before=None, after=None):
    """Stream git history policy changes to destination.


    Default stream destination is a summary of the policy changes to stdout, one
    per line. Also supported for stdout streaming is `jsonline`.

    AWS Kinesis and SQS destinations are specified by providing the ARN.

    Database destinations are supported by providing a sqlalchemy DSN. Note
    SQLAlchemy and db drivers must be installed separately as they an optional
    dependency.

    When using database destinations, streaming defaults to incremental.
    """
    logging.basicConfig(
        format="%(asctime)s: %(name)s:%(levelname)s %(message)s",
        level=(verbose and logging.DEBUG or logging.INFO))
    logging.getLogger('botocore').setLevel(logging.WARNING)

    if before:
        before = parse(before)
    if after:
        after = parse(after)
    if sort:
        sort = six.moves.reduce(operator.or_, [SORT_TYPE[s] for s in sort])

    with contextlib.closing(TempDir().open()) as temp_dir:
        if repo_uri is None:
            repo_uri = pygit2.discover_repository(os.getcwd())
            log.debug("Using repository %s", repo_uri)
        if repo_uri.startswith('http') or repo_uri.startswith('git@'):
            log.info("Cloning repository: %s", repo_uri)
            repo = pygit2.clone_repository(repo_uri, temp_dir.path)
        else:
            repo = pygit2.Repository(repo_uri)
        load_resources()
        policy_repo = PolicyRepo(repo_uri, repo)
        change_count = 0

        with contextlib.closing(transport(stream_uri, assume)) as t:
            if after is None and isinstance(t, IndexedTransport):
                after = t.last()
            for change in policy_repo.delta_stream(after=after, before=before):
                change_count += 1
                t.send(change)

        log.info("Streamed %d policy repo changes", change_count)
    return change_count