collect

Quick start guide for CivicLens' internal library for accessing data via the regulations.gov API. This library contains methods for collecting comments, dockets, and documents from regulations.gov.

Getting Started

To collect data from regulations.gov, you'll first need an API key which you can request here. Once you have your key, you can add it as an Environment Variable by following the steps described under Contributing, or you can replace api_key with your unique key.

Example 1 - Retrieving data from one docket

Let's walk through how to collect data for the proposed rule "FR-6362-P-01 Reducing Barriers to HUD-Assisted Housing." You can read the actual rule and view comments at this link. To collect data on this rule, we'll need to search by the Document ID, which we can pass into the params argument like this:

from access_api_data import pull_reg_gov_data

api_key = "YOUR-KEY_HERE"
search_terms = {"filter[searchTerm]": "HUD-2024-0031-0001"}

doc = pull_reg_gov_data(
    api_key,
    data_type="documents",  # set the datatype to document
    params=search_terms,
)

Now that we've collected the metadata for the document, we can use the documents's object ID to collect all the comments posted to it.

doc_object_id = doc[0]["attributes"]["objectId"]

comment_data = pull_reg_gov_data(
    api_key,
    data_type="comments",
    params={"filter[commentOnId]": doc_object_id},
)

Finally, we can get the text for each comment by iterating over the comment metadata and making an additional request for the text.

comment_json_text = []
for commentId in comment_data[0]["attributes"]["objectId"]:
    comment_json_text.append(
        pull_reg_gov_data(
            api_key,
            data_type="comments",
            params={"filter[commentOnId]": commentId},
        )
    )

Example 2 - Retrieving all comments made in the first two weeks of April 2024

In this example, we demonstrate how to gather all public comments posted within the first ten days of April 2024. This method can also be applied to collect documents or dockets by adjusting 'data_type'.

comments_apr_01_10 = pull_reg_gov_data(
    api_key,
    data_type="comments",
    start_date="2024-05-01",
    end_date="2024-05-10",
    )

Reference

access_api_data

This code pulls heavily from the following existing repositories:

https://github.com/willjobs/regulations-public-comments https://github.com/jacobfeldgoise/regulations-comments-downloader

pull_reg_gov_data(api_key, data_type, start_date=None, end_date=None, params=None, print_remaining_requests=False, skip_duplicates=False)

Returns the JSON associated with a request to the API; max length of 24000

Draws heavily from this [repository] (https://github.com/willjobs/regulations-public-comments/blob/master/ comments_downloader.py)

Parameters:

Name	Type	Description	Default
data_type	str	'dockets', 'documents', or 'comments' -- what kind of data we want back from the API	required
start_date	str in YYYY-MM-DD format	the inclusive start date of our data pull	None
end_date	str in YYYY-MM-DD format	the inclusive end date of our data pull	None
params	dict	Parameters to specify to the endpoint request. Defaults to None. If we are querying the non-details endpoint, we also append the "page[size]" parameter so that we always get the maximum page size of 250 elements per page.	None
print_remaining_requests	bool	Whether to print out the number of remaining requests this hour, based on the response headers. Defaults to False.	False
wait_for_rate_reset	bool	Determines whether to wait to re-try if we run out of requests in a given hour. Defaults to False.	required
skip_duplicates	bool	If a request returns multiple items when only 1 was expected, should we skip that request? Defaults to False.	False

Returns:

Name	Type	Description
dict		JSON-ified request response

Source code in civiclens/collect/access_api_data.py

def pull_reg_gov_data(  # noqa: C901,E501
    api_key,
    data_type,
    start_date=None,
    end_date=None,
    params=None,
    print_remaining_requests=False,
    skip_duplicates=False,
):
    """
    Returns the JSON associated with a request to the API; max length of 24000

    Draws heavily from this [repository]
    (https://github.com/willjobs/regulations-public-comments/blob/master/
    comments_downloader.py)

    Args:
        data_type (str): 'dockets', 'documents', or 'comments' -- what kind of
            data we want back from the API
        start_date (str in YYYY-MM-DD format, optional): the inclusive start
            date of our data pull
        end_date (str in YYYY-MM-DD format, optional): the inclusive end date
            of our data pull
        params (dict, optional): Parameters to specify to the endpoint request.
            Defaults to None.
            If we are querying the non-details endpoint, we also append the
            "page[size]" parameter so that we always get the maximum page size
            of 250 elements per page.
        print_remaining_requests (bool, optional): Whether to print out the
            number of remaining requests this hour, based on the response
            headers.
            Defaults to False.
        wait_for_rate_reset (bool, optional): Determines whether to wait to
            re-try if we run out of requests in a given hour. Defaults to
            False.
        skip_duplicates (bool, optional): If a request returns multiple items
            when only 1 was expected, should we skip that request? Defaults to
            False.

    Returns:
        dict: JSON-ified request response
    """
    # generate the right API request
    api_url = "https://api.regulations.gov/v4/"
    endpoint = f"{api_url}{data_type}"
    params = params if params is not None else {}

    # Our API key has a rate limit of 1,000 requests/hour.
    # If we hit that limit, we can
    # retry every WAIT_MINUTES minutes (more frequently than once an hour, in
    # case our request limit
    # is updated sooner). We will sleep for POLL_SECONDS seconds at a time to
    # see if we've been
    # interrupted. Otherwise we'd have to wait a while before getting
    # interrupted. We could do this
    # with threads, but that gets more complicated than it needs to be.
    STATUS_CODE_OVER_RATE_LIMIT = 429
    WAIT_SECONDS = 3600  # Default to 1 hour

    # if any dates are specified, format those and add to the params
    if start_date or end_date:
        param_dates = api_date_format_params(start_date, end_date)
        params.update(param_dates)

    # whether we are querying the search endpoint (e.g., /documents)
    # or the "details" endpoint
    if endpoint.split("/")[-1] in ["dockets", "documents", "comments"]:
        params = {**params, "page[size]": 250}  # always get max page size

    # Rather than do requests.get(), use this approach to (attempt to)
    # gracefully handle noisy connections to the server
    # We sometimes get SSL errors (unexpected EOF or ECONNRESET), so this
    # should hopefully help us retry.
    session = requests.Session()
    session.mount("https", HTTPAdapter(max_retries=4))

    def poll_for_response(api_key, wait_for_rate_reset):
        r = session.get(
            endpoint, headers={"X-Api-Key": api_key}, params=params, verify=True
        )

        if r.status_code == 200:
            # SUCCESS! Return the JSON of the request
            num_requests_left = int(r.headers["X-RateLimit-Remaining"])
            if (
                print_remaining_requests
                or (num_requests_left < 10)
                or (num_requests_left <= 100 and num_requests_left % 10 == 0)
                or (num_requests_left % 100 == 0 and num_requests_left < 1000)
            ):
                print(f"(Requests left: {r.headers['X-RateLimit-Remaining']})")

            return [True, r.json()]
        else:
            if (
                r.status_code == STATUS_CODE_OVER_RATE_LIMIT
                and wait_for_rate_reset
            ):
                the_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                retry_after = r.headers.get("Retry-After", None)
                wait_time = (
                    int(retry_after)
                    if retry_after and retry_after.isdigit()
                    else WAIT_SECONDS
                )
                print(
                    f"""Rate limit exceeded at {the_time}.
                    Waiting {wait_time} seconds to retry."""
                )
                time.sleep(wait_time)
            elif _is_duplicated_on_server(r.json()) and skip_duplicates:
                print("****Duplicate entries on server. Skipping.")
                print(r.json()["errors"][0]["detail"])
            else:  # some other kind of error
                print([r, r.status_code])
                print(r.json())
                r.raise_for_status()

        return [False, r.json()]

    if data_type == "comments" or data_type == "documents":
        all_objects = []
        unique_objects = {}

        params.update(
            {
                "page[size]": 250,
                "sort": "lastModifiedDate,documentId",
            }
        )

        last_modified_date = None
        continue_fetching = True

        while continue_fetching:
            success, r_json = poll_for_response(
                api_key, wait_for_rate_reset=True
            )
            if success:
                all_objects.extend(r_json["data"])
                print(
                    f"""Fetched {len(r_json['data'])} objects,
                    total: {len(all_objects)}"""
                )

                # Check and handle the pagination
                has_next_page = r_json["meta"].get("hasNextPage", False)
                print(f"Has next page: {has_next_page}")

                if len(r_json["data"]) < 250 or not has_next_page:
                    print(
                        """No more pages or fewer than 250
                        comments fetched, stopping..."""
                    )
                    continue_fetching = False
                else:
                    last_modified_date = format_datetime_for_api(
                        r_json["data"][-1]["attributes"]["lastModifiedDate"]
                    )
                    params.update(
                        {
                            "filter[lastModifiedDate][ge]": last_modified_date,
                            "page[size]": 250,
                            "sort": "lastModifiedDate,documentId",
                            "page[number]": 1,
                            "api_key": api_key,
                        }
                    )
                    if end_date:
                        params.update(
                            {
                                "filter[lastModifiedDate][le]": f"{end_date} 23:59:59"  # noqa: E501
                            }
                        )
                    print(f"Fetching more data from {last_modified_date}")
            else:
                print("Failed to fetch data")
                continue_fetching = False

        # Remove Duplicates
        for obj in all_objects:
            unique_objects[obj["id"]] = obj
        return list(unique_objects.values())

    else:
        doc_data = None  # Initialize doc_data to None
        for i in range(1, 21):  # Fetch up to 20 pages
            params.update(
                {
                    "page[size]": 250,
                    # Ensure that only lastModifiedDate is considered,
                    # dockets cant take in documentID
                    "sort": "lastModifiedDate",
                    "page[number]": str(i),
                    "api_key": api_key,
                }
            )

            success, r_json = poll_for_response(
                api_key, wait_for_rate_reset=True
            )

            if success or (
                _is_duplicated_on_server(r_json) and skip_duplicates
            ):
                if doc_data is not None:
                    doc_data += r_json["data"]
                else:
                    doc_data = r_json["data"]

            # Break if it's the last page
            if r_json["meta"]["lastPage"]:
                return doc_data

    raise RuntimeError(f"Unrecoverable error; {r_json}")

api_date_format_params(start_date=None, end_date=None)

Formats dates to be passed to API call. Assumes we want whole days, and aren't filtering by time.

Parameters:

Name	Type	Description	Default
start_date	str in YYYY-MM-DD format	the inclusive start date of our data pull	None
end_date	str in YYYY-MM-DD format	the inclusive end date of our data pull	None

Returns:

Name	Type	Description
date_param	dict	dict containing the right formatted date calls

Source code in civiclens/collect/access_api_data.py

def api_date_format_params(start_date=None, end_date=None):
    """
    Formats dates to be passed to API call. Assumes we want whole days, and
    aren't filtering by time.

    Args:
        start_date (str in YYYY-MM-DD format, optional): the inclusive start
            date of our data pull
        end_date (str in YYYY-MM-DD format, optional): the inclusive end date
            of our data pull

    Returns:
        date_param (dict): dict containing the right formatted date calls
    """
    date_param = {}
    if start_date is not None:
        date_param.update(
            {"filter[lastModifiedDate][ge]": f"{start_date} 00:00:00"}
        )
        print(f"{start_date=}")
    if end_date is not None:
        date_param.update(
            {"filter[lastModifiedDate][le]": f"{end_date} 23:59:59"}
        )
        print(f"{end_date=}")

    return date_param

move_data_from_api_to_database

add_comments_based_on_comment_date_range(start_date, end_date)

Add comments to the comments table based on a date range of when the comments were posted

Parameters:

Name	Type	Description	Default
start_date	str	the date in YYYY-MM-DD format to pull data from (inclusive)	required
end_date	str	the date in YYYY-MM-DD format to stop the data pull (inclusive)	required

Returns: nothing; adds comments, if available, to the db

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_comments_based_on_comment_date_range(
    start_date: str, end_date: str
) -> None:
    """
    Add comments to the comments table based on a date range of when the
    comments were posted

    Args:
        start_date (str): the date in YYYY-MM-DD format to pull data from
            (inclusive)
        end_date (str): the date in YYYY-MM-DD format to stop the data pull
            (inclusive)

    Returns: nothing; adds comments, if available, to the db
    """
    comment_data = pull_reg_gov_data(
        REG_GOV_API_KEY,
        "comments",
        start_date=start_date,
        end_date=end_date,
    )

    for comment in comment_data:
        logging.info(f"processing comment {comment['id']} ")

        all_comment_data = merge_comment_text_and_data(REG_GOV_API_KEY, comment)

        document_id = all_comment_data["data"]["attributes"].get(
            "commentOnDocumentId", ""
        )

        logging.info(f"checking {document_id} is in the db")

        if verify_database_existence(
            "regulations_document",
            document_id,
            "id",
        ):
            logging.info(
                f"{document_id} is in the db! begin processing comment"
            )

            # clean
            clean_comment_data(all_comment_data)

            # qa
            qa_comment_data(all_comment_data)

            # insert
            insert_response = insert_comment_into_db(all_comment_data)
            if insert_response["error"]:
                logging.error(insert_response["description"])
            else:
                logging.info(
                    f"added comment {all_comment_data['id']} to the db"
                )

add_comments_to_db(doc_list, print_statements=True)

Add comments on a list of documents to the database

Parameters:

Name	Type	Description	Default
doc_list	list of json objects	what is returned from an API call for documents	required
print_statements	boolean	whether to print info on progress	True

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_comments_to_db(
    doc_list: list[dict], print_statements: bool = True
) -> None:
    """
    Add comments on a list of documents to the database

    Args:
        doc_list (list of json objects): what is returned from an API call for
            documents
        print_statements (boolean): whether to print info on progress

    """
    for doc in doc_list:
        document_id = doc["id"]
        document_object_id = doc["attributes"]["objectId"]
        commentable = doc["attributes"]["openForComment"]
        # get the comments, comment text, and add to db
        if commentable:
            if not verify_database_existence(
                "regulations_comment", document_id, "document_id"
            ):  # doc doesn't exist in the db; it's new
                if print_statements:
                    logging.info(
                        f"""no comments found in database for document
                        {document_id}"""
                    )

                add_comments_to_db_for_new_doc(document_object_id)

                if print_statements:
                    logging.info(
                        f"""tried to add comments on document
                        {document_id} to the db"""
                    )

            else:  # doc exists in db; only need to add new comments
                add_comments_to_db_for_existing_doc(
                    document_id, document_object_id, print_statements
                )

add_comments_to_db_for_existing_doc(document_id, document_object_id, print_statements=True)

Gets the most recent comment in the comments table and pulls comments for a doc from the API since then

Parameters:

Name	Type	Description	Default
document	id (str	the id for the document we want comments for (comes from the document json object)	required
document_object_id	str	the object id for the document we want comments for (comes from the document json object)	required
print_statements	bool	whether to print during, default is true	True

Returns: nothing; adds comments, if available, to the db

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_comments_to_db_for_existing_doc(
    document_id: str, document_object_id: str, print_statements: bool = True
) -> None:
    """
    Gets the most recent comment in the comments table and pulls comments for
        a doc from the API since then

    Args:
        document id (str): the id for the document we want comments for (comes
            from the document json object)
        document_object_id (str): the object id for the document we want
            comments for (comes from the document json object)
        print_statements (bool): whether to print during, default is true

    Returns: nothing; adds comments, if available, to the db
    """

    # get date of most recent comment on this doc in the db
    most_recent_comment_date = get_most_recent_doc_comment_date(document_id)
    if print_statements:
        logging.info(
            f"""comments found for document {document_id}, most recent was
            {most_recent_comment_date}"""
        )

    comment_data = pull_reg_gov_data(
        REG_GOV_API_KEY,
        "comments",
        params={
            "filter[commentOnId]": document_object_id,
            "filter[lastModifiedDate][ge]": most_recent_comment_date,
        },
    )

    for comment in comment_data:
        all_comment_data = merge_comment_text_and_data(REG_GOV_API_KEY, comment)

        # if documumrnent is not in the db, add it
        document_id = all_comment_data.get("commentOnDocumentId", "")
        if not verify_database_existence(
            "regulations_document", document_id, "id"
        ):
            # call the API to get the document info
            doc_list = pull_reg_gov_data(
                REG_GOV_API_KEY,
                "documents",
                params={"filter[objectId]": document_id},
            )
            add_documents_to_db(doc_list)

        # clean
        clean_comment_data(all_comment_data)

        # qa
        qa_comment_data(all_comment_data)

        insert_response = insert_comment_into_db(all_comment_data)
        if insert_response["error"]:
            logging.error(insert_response["description"])

add_comments_to_db_for_new_doc(document_object_id)

Add comments to the comments table for a new doc (ie, when we have just added the doc to the database)

Parameters:

Name	Type	Description	Default
document_object_id	str	the object id for the document we want comments for (comes from the document json object)	required

Returns: nothing; adds comments, if available, to the db

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_comments_to_db_for_new_doc(document_object_id: str) -> None:
    """
    Add comments to the comments table for a new doc (ie, when we have just
        added the doc to the database)

    Args:
        document_object_id (str): the object id for the document we want
            comments for (comes from the document json object)

    Returns: nothing; adds comments, if available, to the db
    """
    comment_data = pull_reg_gov_data(
        REG_GOV_API_KEY,
        "comments",
        params={"filter[commentOnId]": document_object_id},
    )
    # add comment data to comments table in the database
    for comment in comment_data:
        all_comment_data = merge_comment_text_and_data(REG_GOV_API_KEY, comment)

        # clean
        clean_comment_data(all_comment_data)

        # qa
        qa_comment_data(all_comment_data)

        insert_response = insert_comment_into_db(all_comment_data)

        if insert_response["error"]:
            logging.error(insert_response["description"])

add_data_quality_flag(data_id, data_type, error_message)

Add data to the regulations_dataqa table is qa assert statements flag

Parameters:

Name	Type	Description	Default
data_id	str	id field of the data in question	required
data_type	str	whether docket, document, or comment	required
error_message	Error	the error and message raised by the assert statement	required

Returns: nothing, add a row to regulations_dataqa

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_data_quality_flag(
    data_id: str, data_type: str, error_message: str
) -> None:
    """Add data to the regulations_dataqa table is qa assert statements flag

    Args:
        data_id (str): id field of the data in question
        data_type (str): whether docket, document, or comment
        error_message (Error): the error and message raised by the assert
            statement
    Returns: nothing, add a row to regulations_dataqa
    """
    connection, cursor = connect_db_and_get_cursor()
    with connection:
        with cursor:
            cursor.execute(
                """INSERT INTO regulations_dataqa (
                    "data_id",
                    "data_type",
                    "error_message",
                    "added_at"
                ) VALUES (%s, %s, %s, %s)
                ON CONFLICT (id) DO UPDATE SET
                    data_id = EXCLUDED.data_id,
                    data_type = EXCLUDED.data_type,
                    error_message = EXCLUDED.error_message,
                    added_at = EXCLUDED.added_at;""",
                (data_id, data_type, str(error_message), datetime.now()),
            )
        connection.commit()

    if connection:
        cursor.close()
        connection.close()

    logging.info(f"Added data quality flag for {data_id} of type {data_type}.")

add_dockets_to_db(doc_list, print_statements=True)

Add the dockets connected to a list of documents into the database

Parameters:

Name	Type	Description	Default
doc_list	list of json objects	what is returned from an API call for documents	required
print_statements	boolean	whether to print info on progress	True

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_dockets_to_db(
    doc_list: list[dict], print_statements: bool = True
) -> None:
    """
    Add the dockets connected to a list of documents into the database

    Args:
        doc_list (list of json objects): what is returned from an API call
            for documents
        print_statements (boolean): whether to print info on progress

    """
    for doc in doc_list:
        docket_id = doc["attributes"]["docketId"]
        commentable = doc["attributes"]["openForComment"]
        # if docket not in db, add it
        if commentable and (
            not verify_database_existence("regulations_docket", docket_id)
        ):
            docket_data = pull_reg_gov_data(
                REG_GOV_API_KEY,
                "dockets",
                params={"filter[searchTerm]": docket_id},
            )

            # clean
            clean_docket_data(docket_data)

            # qa
            qa_docket_data(docket_data)

            # add docket_data to docket table in the database
            insert_response = insert_docket_into_db(docket_data)
            if insert_response["error"]:
                logging.error(insert_response["description"])
            else:
                if print_statements:
                    logging.info(f"Added docket {docket_id} to the db")

add_documents_to_db(doc_list, print_statements=True)

Add a list of document json objects into the database

Parameters:

Name	Type	Description	Default
doc_list	list of json objects	what is returned from an API call for documents	required
print_statements	boolean	whether to print info on progress	True

Source code in civiclens/collect/move_data_from_api_to_database.py

def add_documents_to_db(
    doc_list: list[dict], print_statements: bool = True
) -> None:
    """
    Add a list of document json objects into the database

    Args:
        doc_list (list of json objects): what is returned from an API call for
            documents
        print_statements (boolean): whether to print info on progress

    """
    for doc in doc_list:
        document_id = doc["id"]
        commentable = doc["attributes"]["openForComment"]
        if commentable and (
            not verify_database_existence("regulations_document", document_id)
        ):
            # add this doc to the documents table in the database
            full_doc_info = query_register_API_and_merge_document_data(doc)
            # qa step
            qa_document_data(full_doc_info)

            # clean step
            clean_document_data(full_doc_info)

            # insert step
            insert_response = insert_document_into_db(full_doc_info)
            if insert_response["error"]:
                logging.info(insert_response["description"])
                # would want to add logging here
            else:
                if print_statements:
                    logging.info(f"added document {document_id} to the db")

check_CFR_data(document_data)

Check that the CFR field looks right

Source code in civiclens/collect/move_data_from_api_to_database.py

def check_CFR_data(document_data: json) -> bool:
    """
    Check that the CFR field looks right
    """
    try:
        assert (
            document_data["CFR"] is None
            or document_data["CFR"] == "Not Found"
            or "CFR" in document_data["CFR"]
            or document_data["CFR"].isalpha()
        )
        return True
    except AssertionError:
        return False

clean_comment_data(comment_data)

Clean comment text -- make sure dates are formatted correctly

Source code in civiclens/collect/move_data_from_api_to_database.py

def clean_comment_data(comment_data: json) -> None:
    """
    Clean comment text -- make sure dates are formatted correctly
    """

    comment_text_attributes = comment_data["data"]["attributes"]

    # format the date fields
    for date_field in ["modifyDate", "postedDate", "receiveDate"]:
        comment_text_attributes[date_field] = (
            datetime.strptime(
                comment_text_attributes[date_field], "%Y-%m-%dT%H:%M:%SZ"
            )
            if comment_text_attributes[date_field]
            else None
        )

    # clean the text
    if comment_text_attributes["comment"] is not None:
        comment_text_attributes["comment"] = clean_text(
            comment_text_attributes["comment"]
        )

clean_document_data(document_data)

Clean document data in place; run cleaning code on summary

Source code in civiclens/collect/move_data_from_api_to_database.py

def clean_document_data(document_data: json) -> None:
    """
    Clean document data in place; run cleaning code on summary
    """
    if document_data["summary"] is not None:
        document_data["summary"] = clean_text(document_data["summary"])

connect_db_and_get_cursor()

Connect to the CivicLens database and return the objects

Source code in civiclens/collect/move_data_from_api_to_database.py

def connect_db_and_get_cursor() -> (
    tuple[psycopg2.extensions.connection, psycopg2.extensions.cursor]
):
    """
    Connect to the CivicLens database and return the objects
    """
    connection = psycopg2.connect(
        database=DATABASE_NAME,
        user=DATABASE_USER,
        password=DATABASE_PASSWORD,
        host=DATABASE_HOST,
        port=DATABASE_PORT,
    )
    cursor = connection.cursor()
    return connection, cursor

extract_xml_text_from_doc(doc)

Take a document's json object, pull the xml text, add the text to the object

Parameters:

Name	Type	Description	Default
doc	json	the object from regulations.gov API	required

Returns:

Name	Type	Description
processed_data	json	the object with added text

Source code in civiclens/collect/move_data_from_api_to_database.py

def extract_xml_text_from_doc(doc: json) -> json:
    """
    Take a document's json object, pull the xml text, add the text to the
    object

    Args:
        doc (json): the object from regulations.gov API

    Returns:
        processed_data (json): the object with added text
    """
    processed_data = []

    if not doc:
        return processed_data

    fr_doc_num = doc["attributes"]["frDocNum"]
    if fr_doc_num:
        xml_url = fetch_fr_document_details(fr_doc_num)
        xml_content = fetch_xml_content(xml_url)
        if xml_content:
            extracted_data = parse_xml_content(xml_content)
            processed_data.append(
                {**doc, **extracted_data}
            )  # merge the json objects

    return processed_data

fetch_fr_document_details(fr_doc_num)

Retrieves xml url for document text from federal register.

Parameters:

Name	Type	Description	Default
fr_doc_num	str	the unique id (comes from regulations.gov api info)	required

Returns:

Type	Description
str	xml url (str)

Source code in civiclens/collect/move_data_from_api_to_database.py

def fetch_fr_document_details(fr_doc_num: str) -> str:
    """
    Retrieves xml url for document text from federal register.

    Args:
        fr_doc_num (str): the unique id (comes from regulations.gov api info)

    Returns:
        xml url (str)
    """
    api_endpoint = f"https://www.federalregister.gov/api/v1/documents/{fr_doc_num}.json?fields[]=full_text_xml_url"  # noqa: E231
    response = requests.get(api_endpoint)
    if response.status_code == 200:
        data = response.json()
        return data.get("full_text_xml_url")
    else:
        error_message = f"Error fetching FR document details for {fr_doc_num}: {response.status_code}"  # noqa: E501
        raise Exception(error_message)

fetch_xml_content(url)

Fetches XML content from a given URL.

Parameters:

Name	Type	Description	Default
url	str	the xml url that we want to retrive text from	required

Returns:

Type	Description
str	response.text (str): the text

Source code in civiclens/collect/move_data_from_api_to_database.py

def fetch_xml_content(url: str) -> str:
    """
    Fetches XML content from a given URL.

    Args:
        url (str): the xml url that we want to retrive text from

    Returns:
        response.text (str): the text
    """
    response = requests.get(url)
    if response.status_code == 200:
        return response.text
    else:
        error_message = (
            f"Error fetching XML content from {url}: {response.status_code}"
        )
        raise Exception(error_message)

get_comment_text(api_key, comment_id)

Get the text of a comment, with retry logic to handle rate limits.

Parameters:

Name	Type	Description	Default
api_key	str	key for the regulations.gov API	required
comment_id	str	the id for the comment	required

Returns:

Type	Description
dict	json object for the comment text

Source code in civiclens/collect/move_data_from_api_to_database.py

def get_comment_text(api_key: str, comment_id: str) -> dict:
    """
    Get the text of a comment, with retry logic to handle rate limits.

    Args:
        api_key (str): key for the regulations.gov API
        comment_id (str): the id for the comment

    Returns:
        json object for the comment text
    """
    api_url = "https://api.regulations.gov/v4/comments/"
    endpoint = f"{api_url}{comment_id}?include=attachments"

    STATUS_CODE_OVER_RATE_LIMIT = 429
    WAIT_SECONDS = 3600  # Default to 1 hour

    session = requests.Session()
    session.mount("https", HTTPAdapter(max_retries=4))

    while True:
        response = session.get(
            endpoint, headers={"X-Api-Key": api_key}, verify=True
        )

        if response.status_code == 200:
            # SUCCESS! Return the JSON of the request
            return response.json()
        elif response.status_code == STATUS_CODE_OVER_RATE_LIMIT:
            retry_after = response.headers.get("Retry-After", None)
            wait_time = (
                int(retry_after)
                if retry_after and retry_after.isdigit()
                else WAIT_SECONDS
            )
            the_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            logging.info(
                f"""Rate limit exceeded at {the_time}.
                Waiting {wait_time} seconds to retry."""
            )
            time.sleep(wait_time)
        else:
            logging.info(
                f"""Failed to retrieve comment data.
                Status code: {response.status_code}"""
            )
            return None

get_most_recent_doc_comment_date(doc_id)

Returns the date of the most recent comment for a doc

Parameters:

Name	Type	Description	Default
doc_id	str	the regulations.gov doc id	required

Returns:

Name	Type	Description
response	datetime	the most recent date

Source code in civiclens/collect/move_data_from_api_to_database.py

def get_most_recent_doc_comment_date(doc_id: str) -> str:
    """
    Returns the date of the most recent comment for a doc

    Args:
        doc_id (str): the regulations.gov doc id

    Returns:
        response (datetime): the most recent date
    """
    connection, cursor = connect_db_and_get_cursor()
    with connection:
        with cursor:
            query = f"""SELECT MAX("posted_date")
                FROM regulations_comment
                WHERE "document_id" = '{doc_id}';"""  # noqa: E231, E702
            cursor.execute(query)
            response = cursor.fetchall()

    if connection:
        cursor.close()
        connection.close()

    # format the text
    # it seems that the regulations.gov API returns postedDate rounded to the
    # hour. If we used that naively as the most recent date, we might miss some
    # comments when we pull comments again. By backing up one hour, we trade
    # off some unnecessary API calls for ensuring we don't miss anything
    date_dt = response[0][0]
    one_hour_prior = date_dt - dt.timedelta(hours=1)
    most_recent_date = one_hour_prior.strftime("%Y-%m-%d %H:%M:%S")

    return most_recent_date

insert_comment_into_db(comment_data)

Insert the info on a comment into the PublicComments table

Parameters:

Name	Type	Description	Default
comment_data	json	the comment info from regulations.gov API	required

Returns:

Type	Description
dict	nothing unless an error; adds the info into the table

Source code in civiclens/collect/move_data_from_api_to_database.py

def insert_comment_into_db(comment_data: json) -> dict:
    """
    Insert the info on a comment into the PublicComments table

    Args:
        comment_data (json): the comment info from regulations.gov API

    Returns:
        nothing unless an error; adds the info into the table
    """
    connection, cursor = connect_db_and_get_cursor()

    attributes = comment_data["attributes"]
    comment_text_attributes = comment_data["data"]["attributes"]

    # Map JSON attributes to corresponding table columns
    comment_id = comment_data["id"]
    objectId = attributes.get("objectId", "")
    commentOn = comment_text_attributes.get("commentOn", "")
    document_id = comment_text_attributes.get("commentOnDocumentId", "")
    duplicateComments = comment_text_attributes.get("duplicateComments", 0)
    stateProvinceRegion = comment_text_attributes.get("stateProvinceRegion", "")
    subtype = comment_text_attributes.get("subtype", "")
    comment = comment_text_attributes.get("comment", "")
    firstName = comment_text_attributes.get("firstName", "")
    lastName = comment_text_attributes.get("lastName", "")
    address1 = comment_text_attributes.get("address1", "")
    address2 = comment_text_attributes.get("address2", "")
    city = comment_text_attributes.get("city", "")
    category = comment_text_attributes.get("category", "")
    country = comment_text_attributes.get("country", "")
    email = comment_text_attributes.get("email", "")
    phone = comment_text_attributes.get("phone", "")
    govAgency = comment_text_attributes.get("govAgency", "")
    govAgencyType = comment_text_attributes.get("govAgencyType", "")
    organization = comment_text_attributes.get("organization", "")
    originalDocumentId = comment_text_attributes.get("originalDocumentId", "")
    modifyDate = comment_text_attributes.get("modifyDate", "")
    pageCount = comment_text_attributes.get("pageCount", 0)
    postedDate = comment_text_attributes.get("postedDate", "")
    receiveDate = comment_text_attributes.get("receiveDate", "")
    title = attributes.get("title", "")
    trackingNbr = comment_text_attributes.get("trackingNbr", "")
    withdrawn = comment_text_attributes.get("withdrawn", False)
    reasonWithdrawn = comment_text_attributes.get("reasonWithdrawn", "")
    zip = comment_text_attributes.get("zip", "")
    restrictReason = comment_text_attributes.get("restrictReason", "")
    restrictReasonType = comment_text_attributes.get("restrictReasonType", "")
    submitterRep = comment_text_attributes.get("submitterRep", "")
    submitterRepAddress = comment_text_attributes.get("submitterRepAddress", "")
    submitterRepCityState = comment_text_attributes.get(
        "submitterRepCityState", ""
    )

    # SQL INSERT statement
    query = """
    INSERT INTO regulations_comment (
        "id",
        "object_id",
        "comment_on",
        "document_id",
        "duplicate_comments",
        "state_province_region",
        "subtype",
        "comment",
        "first_name",
        "last_name",
        "address1",
        "address2",
        "city",
        "category",
        "country",
        "email",
        "phone",
        "gov_agency",
        "gov_agency_type",
        "organization",
        "original_document_id",
        "modify_date",
        "page_count",
        "posted_date",
        "receive_date",
        "title",
        "tracking_nbr",
        "withdrawn",
        "reason_withdrawn",
        "zip",
        "restrict_reason",
        "restrict_reason_type",
        "submitter_rep",
        "submitter_rep_address",
        "submitter_rep_city_state"
    ) VALUES (
        %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
        %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s
    )
    ON CONFLICT (id) DO UPDATE SET
        object_id = EXCLUDED.object_id,
        comment_on = EXCLUDED.comment_on,
        document_id = EXCLUDED.document_id,
        duplicate_comments = EXCLUDED.duplicate_comments,
        state_province_region = EXCLUDED.state_province_region,
        subtype = EXCLUDED.subtype,
        comment = EXCLUDED.comment,
        first_name = EXCLUDED.first_name,
        last_name = EXCLUDED.last_name,
        address1 = EXCLUDED.address1,
        address2 = EXCLUDED.address2,
        city = EXCLUDED.city,
        category = EXCLUDED.category,
        country = EXCLUDED.country,
        email = EXCLUDED.email,
        phone = EXCLUDED.phone,
        gov_agency = EXCLUDED.gov_agency,
        gov_agency_type = EXCLUDED.gov_agency_type,
        organization = EXCLUDED.organization,
        original_document_id = EXCLUDED.original_document_id,
        modify_date = EXCLUDED.modify_date,
        page_count = EXCLUDED.page_count,
        posted_date = EXCLUDED.posted_date,
        receive_date = EXCLUDED.receive_date,
        title = EXCLUDED.title,
        tracking_nbr = EXCLUDED.tracking_nbr,
        withdrawn = EXCLUDED.withdrawn,
        reason_withdrawn = EXCLUDED.reason_withdrawn,
        zip = EXCLUDED.zip,
        restrict_reason = EXCLUDED.restrict_reason,
        restrict_reason_type = EXCLUDED.restrict_reason_type,
        submitter_rep = EXCLUDED.submitter_rep,
        submitter_rep_address = EXCLUDED.submitter_rep_address,
        submitter_rep_city_state = EXCLUDED.submitter_rep_city_state;
    """

    # Execute the SQL statement
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                query,
                (
                    comment_id,
                    objectId,
                    commentOn,
                    document_id,
                    duplicateComments,
                    stateProvinceRegion,
                    subtype,
                    comment,
                    firstName,
                    lastName,
                    address1,
                    address2,
                    city,
                    category,
                    country,
                    email,
                    phone,
                    govAgency,
                    govAgencyType,
                    organization,
                    originalDocumentId,
                    modifyDate,
                    pageCount,
                    postedDate,
                    receiveDate,
                    title,
                    trackingNbr,
                    withdrawn,
                    reasonWithdrawn,
                    zip,
                    restrictReason,
                    restrictReasonType,
                    submitterRep,
                    submitterRepAddress,
                    submitterRepCityState,
                ),
            )
            connection.commit()

        if connection:
            connection.close()

    except Exception as e:
        error_message = f"""Error inserting comment {comment_data['id']} into
        comment table: {e}"""
        logging.error(error_message)
        return {
            "error": True,
            "message": e,
            "description": error_message,
        }

    return {
        "error": False,
        "message": None,
        "description": None,
    }

insert_docket_into_db(docket_data)

Run assert statements to check docket data looks right

Parameters:

Name	Type	Description	Default
docket_data	json	the docket info from regulations.gov API	required

Returns: nothing unless an error; adds the info into the table

Source code in civiclens/collect/move_data_from_api_to_database.py

def insert_docket_into_db(docket_data: json) -> dict:
    """
    Run assert statements to check docket data looks right

    Args:
        docket_data (json): the docket info from regulations.gov API

    Returns: nothing unless an error; adds the info into the table
    """

    data_for_db = docket_data[0]
    attributes = data_for_db["attributes"]
    try:
        connection, cursor = connect_db_and_get_cursor()

        with connection:
            with cursor:
                cursor.execute(
                    """
                    INSERT INTO regulations_docket (
                        "id",
                        "docket_type",
                        "last_modified_date",
                        "agency_id",
                        "title",
                        "object_id",
                        "highlighted_content"
                    ) VALUES (
                        %s, %s, %s, %s, %s, %s, %s
                    )
                    ON CONFLICT (id) DO UPDATE SET
                        docket_type = EXCLUDED.docket_type,
                        last_modified_date = EXCLUDED.last_modified_date,
                        agency_id = EXCLUDED.agency_id,
                        title = EXCLUDED.title,
                        object_id = EXCLUDED.object_id,
                        highlighted_content = EXCLUDED.highlighted_content;
                    """,
                    (
                        data_for_db["id"],
                        attributes["docketType"],
                        attributes["lastModifiedDate"],
                        attributes["agencyId"],
                        attributes["title"],
                        attributes["objectId"],
                        attributes["highlightedContent"],
                    ),
                )
                connection.commit()
    except Exception as e:
        error_message = f"""Error inserting docket {data_for_db["id"]}
        into dockets table: {e}"""
        logging.error(error_message)
        return {
            "error": True,
            "message": e,
            "description": error_message,
        }

    return {
        "error": False,
        "message": None,
        "description": None,
    }

insert_document_into_db(document_data)

Insert the info on a document into the documents table

Parameters:

Name	Type	Description	Default
document_data	json	the document info from regulations.gov API	required

Returns:

Type	Description
dict	nothing unless an error; adds the info into the table

Source code in civiclens/collect/move_data_from_api_to_database.py

def insert_document_into_db(document_data: json) -> dict:
    """
    Insert the info on a document into the documents table

    Args:
        document_data (json): the document info from regulations.gov API

    Returns:
        nothing unless an error; adds the info into the table
    """
    attributes = document_data["attributes"]

    fields_to_insert = (
        document_data["id"],
        attributes["documentType"],
        attributes["lastModifiedDate"],
        attributes["frDocNum"],
        attributes["withdrawn"],
        attributes["agencyId"],
        attributes["commentEndDate"],
        attributes["postedDate"],
        attributes["docketId"],
        attributes["subtype"],
        attributes["commentStartDate"],
        attributes["openForComment"],
        attributes["objectId"],
        document_data["links"]["self"],
        document_data["agencyType"],
        document_data["CFR"],
        document_data["RIN"],
        attributes["title"],
        document_data["summary"],
        document_data["dates"],
        document_data["furtherInformation"],
        document_data["supplementaryInformation"],
    )

    query = """INSERT INTO regulations_document ("id",
                            "document_type",
                            "last_modified_date",
                            "fr_doc_num",
                            "withdrawn",
                            "agency_id",
                            "comment_end_date",
                            "posted_date",
                            "docket_id",
                            "subtype",
                            "comment_start_date",
                            "open_for_comment",
                            "object_id",
                            "full_text_xml_url",
                            "agency_type",
                            "cfr",
                            "rin",
                            "title",
                            "summary",
                            "dates",
                            "further_information",
                            "supplementary_information") \
                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                        %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    ON CONFLICT (id) DO UPDATE SET
                        document_type = EXCLUDED.document_type,
                        last_modified_date = EXCLUDED.last_modified_date,
                        fr_doc_num = EXCLUDED.fr_doc_num,
                        withdrawn = EXCLUDED.withdrawn,
                        agency_id = EXCLUDED.agency_id,
                        comment_end_date = EXCLUDED.comment_end_date,
                        posted_date = EXCLUDED.posted_date,
                        docket_id = EXCLUDED.docket_id,
                        subtype = EXCLUDED.subtype,
                        comment_start_date = EXCLUDED.comment_start_date,
                        open_for_comment = EXCLUDED.open_for_comment,
                        object_id = EXCLUDED.object_id,
                        full_text_xml_url = EXCLUDED.full_text_xml_url,
                        agency_type = EXCLUDED.agency_type,
                        cfr = EXCLUDED.cfr,
                        rin = EXCLUDED.rin,
                        title = EXCLUDED.title,
                        summary = EXCLUDED.summary,
                        dates = EXCLUDED.dates,
                        further_information = EXCLUDED.further_information,
                        supplementary_information =
                        EXCLUDED.supplementary_information;"""
    try:
        connection, cursor = connect_db_and_get_cursor()
        with connection:
            with cursor:
                cursor.execute(
                    query,
                    fields_to_insert,
                )
                connection.commit()
        if connection:
            cursor.close()
            connection.close()

    except Exception as e:
        error_message = f"""Error inserting document
        {document_data['id']} into dockets table: {e}"""
        logging.error(error_message)
        return {
            "error": True,
            "message": e,
            "description": error_message,
        }

    return {
        "error": False,
        "message": None,
        "description": None,
    }

merge_comment_text_and_data(api_key, comment_data)

Combine comment json object with the comment text json object

Parameters:

Name	Type	Description	Default
api_key	str	key for the regulations.gov API	required
comment_data	json	the json object from regulations.gov	required

Returns:

Type	Description
json	combined json object for the comment and text

Source code in civiclens/collect/move_data_from_api_to_database.py

def merge_comment_text_and_data(api_key: str, comment_data: json) -> json:
    """
    Combine comment json object with the comment text json object

    Args:
        api_key (str): key for the regulations.gov API
        comment_data (json): the json object from regulations.gov

    Returns:
        combined json object for the comment and text
    """

    comment_text_data = get_comment_text(api_key, comment_data["id"])

    # DECISION: only track the comment data; don't include the info on comment
    # attachments which is found elsewhere in comment_text_data

    all_comment_data = {**comment_data, **comment_text_data}
    return all_comment_data

parse_xml_content(xml_content)

Parses XML content and extracts relevant data such as agency type, CFR, RIN, title, summary, etc.

Parameters:

Name	Type	Description	Default
xml_content	str	xml formatted text	required

Returns:

Name	Type	Description
extracted_data	dict	contains key parts of the extracted text

Source code in civiclens/collect/move_data_from_api_to_database.py

def parse_xml_content(xml_content: str) -> dict:
    """
    Parses XML content and extracts relevant data such as agency type, CFR,
    RIN, title, summary, etc.

    Args:
        xml_content (str): xml formatted text

    Returns:
        extracted_data (dict): contains key parts of the extracted text
    """
    # Convert the XML string to an ElementTree object
    root = ET.fromstring(xml_content)

    # Initialize a dictionary to hold extracted data
    extracted_data = {}

    # Extract Agency Type
    agency_type = root.find('.//AGENCY[@TYPE="S"]')
    extracted_data["agencyType"] = (
        agency_type.text if agency_type is not None else "Not Found"
    )

    # Extract CFR
    cfr = root.find(".//CFR")
    extracted_data["CFR"] = cfr.text if cfr is not None else "Not Found"

    # Extract RIN
    rin = root.find(".//RIN")
    extracted_data["RIN"] = rin.text if rin is not None else "Not Found"

    # Extract Title (Subject)
    title = root.find(".//SUBJECT")
    extracted_data["title"] = title.text if title is not None else "Not Found"

    # Extract Summary
    summary = root.find(".//SUM/P")
    extracted_data["summary"] = (
        summary.text if summary is not None else "Not Found"
    )

    # Extract DATES
    dates = root.find(".//DATES/P")
    extracted_data["dates"] = dates.text if dates is not None else "Not Found"

    # Extract Further Information
    furinf = root.find(".//FURINF/P")
    extracted_data["furtherInformation"] = (
        furinf.text if furinf is not None else "Not Found"
    )

    # Extract Supplementary Information
    supl_info_texts = []
    supl_info_elements = root.findall(".//SUPLINF/*")
    for element in supl_info_elements:
        # Assuming we want to gather all text from children tags within
        # <SUPLINF>
        if element.text is not None:
            supl_info_texts.append(element.text.strip())
        for sub_element in element:
            if sub_element.text is not None:
                supl_info_texts.append(sub_element.text.strip())

    # Join all pieces of supplementary information text into a single string
    extracted_data["supplementaryInformation"] = " ".join(supl_info_texts)

    return extracted_data

pull_all_api_data_for_date_range(start_date, end_date, pull_dockets, pull_documents, pull_comments)

Pull different types of data from regulations.gov API based on date range

Parameters:

Name	Type	Description	Default
start_date	str	the date in YYYY-MM-DD format to pull data from (inclusive)	required
end_date	str	the date in YYYY-MM-DD format to stop the data pull (inclusive)	required
pull_dockets	boolean		required

Returns:

Type	Description
None	None; adds data to the db

Source code in civiclens/collect/move_data_from_api_to_database.py

def pull_all_api_data_for_date_range(
    start_date: str,
    end_date: str,
    pull_dockets: bool,
    pull_documents: bool,
    pull_comments: bool,
) -> None:
    """
    Pull different types of data from regulations.gov API based on date range

    Args:
        start_date (str): the date in YYYY-MM-DD format to pull data from
            (inclusive)
        end_date (str): the date in YYYY-MM-DD format to stop the data pull
            (inclusive)
        pull_dockets (boolean):

    Returns:
        None; adds data to the db
    """

    # get documents
    logging.info("getting list of documents within date range")
    doc_list = pull_reg_gov_data(
        REG_GOV_API_KEY,
        "documents",
        start_date=start_date,
        end_date=end_date,
    )
    logging.info(f"got {len(doc_list)} documents")
    logging.info("now extracting docs open for comments from that list")

    # pull the commentable docs from that list
    commentable_docs = []
    for doc in doc_list:
        docket_id = doc["attributes"]["docketId"]
        if docket_id and doc["attributes"]["openForComment"]:
            commentable_docs.append(doc)
            # can't add this doc to the db right now because docket needs to be
            # added first
            # the db needs the docket primary key first

    logging.info(f"{len(commentable_docs)} documents open for comment")

    if pull_dockets:
        logging.info(
            "getting dockets associated with the documents open for comment"
        )
        add_dockets_to_db(commentable_docs)
        logging.info("no more dockets to add to db")

    if pull_documents:
        logging.info("adding documents to the db")
        add_documents_to_db(commentable_docs)
        logging.info("no more documents to add to db")

    if pull_comments:
        logging.info("adding comments to the db")
        add_comments_based_on_comment_date_range(start_date, end_date)
        logging.info("no more comments to add to db")

    logging.info("process finished")

qa_comment_data(comment_data)

Run assert statements to check comment data looks right

Parameters:

Name	Type	Description	Default
comment_data	json object	the document data from the API	required

Returns: (bool) whether data is in the expected format

Source code in civiclens/collect/move_data_from_api_to_database.py

def qa_comment_data(comment_data: json) -> None:
    """
    Run assert statements to check comment data looks right

    Args:
        comment_data (json object): the document data from the API

    Returns: (bool) whether data is in the expected format
    """

    attributes = comment_data["attributes"]
    comment_text_attributes = comment_data["data"]["attributes"]
    try:
        assert len(comment_data["id"]) < 255, "id is more than 255 characters"

        assert (
            attributes["objectId"][:2] == "09"
        ), "objectId does not start with '09'"
        assert (
            comment_text_attributes["commentOn"][:2] == "09"
        ), "commentOn does not start with '09'"
        # comment_data["commentOnDocumentId"]
        assert (
            comment_text_attributes["duplicateComments"] == 0
        ), "duplicateComments != 0"
        # assert comment_data["stateProvinceRegion"]
        assert comment_text_attributes[
            "subtype"
        ] is None or comment_text_attributes["subtype"] in [
            "Public Comment",
            "Comment(s)",
        ], "subtype is not an expected value"
        assert isinstance(
            comment_text_attributes["comment"], str
        ), "comment is not string"
        # comment_data["firstName"]
        # comment_data["lastName"]
        # comment_data["address1"]
        # comment_data["address2"]
        # comment_data["city"]
        # comment_data["category"]
        # comment_data["country"]
        # comment_data["email"]
        # comment_data["phone"]
        # comment_data["govAgency"]
        # comment_data["govAgencyType"]
        # comment_data["organization"]
        # comment_data["originalDocumentId"]
        assert isinstance(
            comment_text_attributes["modifyDate"], datetime
        ), "modifyDate is not datetime"
        # comment_data["pageCount"]
        assert isinstance(
            comment_text_attributes["postedDate"], datetime
        ), "postedDate is not datetime"
        assert isinstance(
            comment_text_attributes["receiveDate"], datetime
        ), "receiveDate is not datetime"
        # comment_data["trackingNbr"]
        assert (
            comment_text_attributes["withdrawn"] is False
        ), "withdrawn is not False"
        # comment_data["reasonWithdrawn"]
        # comment_data["zip"]
        # comment_data["restrictReason"]
        # comment_data["restrictReasonType"]
        # comment_data["submitterRep"]
        # comment_data["submitterRepAddress"]
        # comment_data["submitterRepCityState"]

    except AssertionError as e:
        id = comment_data["id"]
        logging.error(f"AssertionError: comment {id} -- {e}")
        add_data_quality_flag(id, "comment", e)

qa_docket_data(docket_data)

Run assert statements to check docket data looks right

Parameters:

Name	Type	Description	Default
docket_data	json object	the docket data from the API	required

Returns: (bool) whether data is in the expected format

Source code in civiclens/collect/move_data_from_api_to_database.py

def qa_docket_data(docket_data: json) -> None:
    """
    Run assert statements to check docket data looks right

    Args:
        docket_data (json object): the docket data from the API

    Returns: (bool) whether data is in the expected format
    """

    attributes = docket_data[0]["attributes"]
    data_for_db = docket_data[0]

    try:
        # need to check that docket_data is in the right format
        assert (
            isinstance(docket_data, list) or len(docket_data) < 1
        ), "docket data in wrong format"

        assert "attributes" in data_for_db, "'attributes' not in docket_data"

        # check the fields
        assert (
            len(data_for_db["id"]) < 255
        ), "id field longer than 255 characters"
        assert attributes["docketType"] in [
            "Rulemaking",
            "Nonrulemaking",
        ], "docketType unexpected value"
        assert (
            len(attributes["lastModifiedDate"]) == 20
            and "202" in attributes["lastModifiedDate"]
        ), "lastModifiedDate is unexpected length"
        assert attributes["agencyId"].isalpha(), "agencyId is not just letter"
        assert isinstance(attributes["title"], str), "title is not string"
        assert (
            attributes["objectId"][:2] == "0b"
        ), "objectId does not start with '0b'"
        # attributes["highlightedContent"]

    except AssertionError as e:
        id = data_for_db["id"]
        logging.error(f"AssertionError: docket {id} -- {e}")
        add_data_quality_flag(id, "docket", e)

qa_document_data(document_data)

Run assert statements to check document data looks right

Parameters:

Name	Type	Description	Default
document_data	json object	the document data from the API	required

Returns: (bool) whether data is in the expected format

Source code in civiclens/collect/move_data_from_api_to_database.py

def qa_document_data(document_data: json) -> True:
    """
    Run assert statements to check document data looks right

    Args:
        document_data (json object): the document data from the API

    Returns: (bool) whether data is in the expected format
    """

    attributes = document_data["attributes"]

    try:
        assert len(document_data["id"]) < 255
        assert attributes["documentType"] in [
            "Proposed Rule",
            "Other",
            "Notice",
            "Not Found",
            "Rule",
        ]
        # attributes["lastModifiedDate"]
        assert validate_fr_doc_num(
            attributes["frDocNum"]
        ), "frDocNum contains unexpected characters or is None"
        assert attributes["withdrawn"] is False, "withdrawn is True"
        # attributes["agencyId"]
        assert (
            len((attributes["commentEndDate"])) == 20
            and "202" in attributes["commentEndDate"]
        ), "commentEndDate is unexpected length"
        assert (
            len(attributes["postedDate"]) == 20
        ), "postedDate is unexpected length"
        # attributes["docketId"]
        # attributes["subtype"]
        assert (
            len(attributes["commentStartDate"]) == 20
            and "202" in attributes["commentStartDate"]
        ), "commentStartDate is expected length"
        assert attributes["openForComment"] is True, "openForComment is False"
        # attributes["objectId"]
        assert (
            "https" in document_data["links"]["self"]
        ), "'https' is not in document_data['links']['self']"
        assert (
            ".gov" in document_data["links"]["self"]
        ), "'.gov' is not in document_data['links']['self']"
        # document_data["agencyType"]
        assert check_CFR_data(document_data), "CFR is not alpha characters"
        # document_data["RIN"]
        assert isinstance(attributes["title"], str), "title is not string"
        assert document_data["summary"] is None or isinstance(
            document_data["summary"], str
        ), "summary is not string"
        # document_data["dates"]
        # document_data["furtherInformation"]
        assert document_data["supplementaryInformation"] is None or isinstance(
            document_data["supplementaryInformation"], str
        ), "supplementaryInformation is not string"

    except AssertionError as e:
        id = document_data["id"]
        logging.error(f"AssertionError: document {id} -- {e}")
        add_data_quality_flag(id, "document", e)

query_register_API_and_merge_document_data(doc)

Attempts to pull document text via federal register API and merge with reg gov API data

Parameters:

Name	Type	Description	Default
doc	json	the raw json for a document from regulations.gov API	required

Returns:

Name	Type	Description
merged_doc	json	the json with fields for text from federal register API

Source code in civiclens/collect/move_data_from_api_to_database.py

def query_register_API_and_merge_document_data(doc: json) -> json:
    """
    Attempts to pull document text via federal register API and merge with reg
    gov API data

    Args:
        doc (json): the raw json for a document from regulations.gov API

    Returns:
        merged_doc (json): the json with fields for text from federal register
            API
    """

    # extract the document text using the general register API
    fr_doc_num = doc.get("attributes", {}).get("frDocNum")
    document_id = doc["id"]
    if fr_doc_num:
        try:
            xml_url = fetch_fr_document_details(fr_doc_num)
            xml_content = fetch_xml_content(xml_url)
            parsed_xml_content = parse_xml_content(xml_content)
            doc.update(parsed_xml_content)  # merge the json objects
        except Exception:
            # if there's an error, that means we can't use the xml_url to get
            # the doc text, so we enter None for those fields
            logging.error(
                rf"""Error accessing federal register xml data for frDocNum
                {fr_doc_num}, \ document id {document_id}"""
            )
            blank_xml_fields = {
                "agencyType": None,
                "CFR": None,
                "RIN": None,
                "title": None,
                "summary": None,
                "dates": None,
                "furtherInformation": None,
                "supplementaryInformation": None,
            }
            doc.update(blank_xml_fields)  # merge the json objects

    else:
        blank_xml_fields = {
            "agencyType": None,
            "CFR": None,
            "RIN": None,
            "title": None,
            "summary": None,
            "dates": None,
            "furtherInformation": None,
            "supplementaryInformation": None,
        }
        doc.update(blank_xml_fields)  # merge the json objects

    return doc

validate_fr_doc_num(field_value)

Check the fr_doc_num field is in the right format

Source code in civiclens/collect/move_data_from_api_to_database.py

def validate_fr_doc_num(field_value):
    """
    Check the fr_doc_num field is in the right format
    """

    # this is a decision: we accept None as a value
    if field_value is None:
        return True

    if field_value == "Not Found":
        return True

    if all(char.isdigit() or char == "-" for char in field_value):
        return True

    # else
    return False

verify_database_existence(table, api_field_val, db_field='id')

Confirm a row exists in a db table for a given id

Parameters:

Name	Type	Description	Default
table	str	one of the tables in the CivicLens db	required
api_field_val	str	the value we're looking for in the table	required
db_field	str	the field in the table where we're looking for the	'id'

Returns:

Type	Description
bool	boolean indicating the value was found

Source code in civiclens/collect/move_data_from_api_to_database.py

def verify_database_existence(
    table: str, api_field_val: str, db_field: str = "id"
) -> bool:
    """
    Confirm a row exists in a db table for a given id

    Args:
        table (str): one of the tables in the CivicLens db
        api_field_val (str): the value we're looking for in the table
        db_field (str): the field in the table where we're looking for the
        value

    Returns:
        boolean indicating the value was found
    """
    connection, cursor = connect_db_and_get_cursor()
    with connection:
        with cursor:
            query = f"SELECT * FROM {table} WHERE {db_field} = %s;"  # noqa: E231, E702
            cursor.execute(query, (api_field_val,))
            response = cursor.fetchall()

    if connection:
        cursor.close()
        connection.close()

    return bool(response)

bulk_dl

BulkDl

Source code in civiclens/collect/bulk_dl.py

class BulkDl:
    def __init__(self, api_key):
        """
        Initializes the BulkDl class.

        Args:
            api_key (str): API key for authenticating requests to the
            regulations.gov API.

        Attributes:
            api_key (str): Stored API key for requests.
            base_url (str): Base URL for the API endpoint.
            headers (dict): Headers to include in API requests, containing API
                key and content type.
            agencies (list[str]): List of agency identifiers (aggregated from
                https://www.regulations.gov/agencies) to be used in API calls.
        """
        self.api_key = api_key
        self.base_url = "https://api.regulations.gov/v4"
        self.headers = {
            "X-Api-Key": self.api_key,
            "Content-Type": "application/json",
        }
        self.agencies = [
            "AID",
            "ATBCB",
            "CFPB",
            "CNCS",
            "COLC",
            "CPPBSD",
            "CPSC",
            "CSB",
            "DHS",
            "DOC",
            "DOD",
            "DOE",
            "DOI",
            "DOJ",
            "DOL",
            "DOS",
            "DOT",
            "ED",
            "EEOC",
            "EIB",
            "EOP",
            "EPA",
            "FFIEC",
            "FMC",
            "FRTIB",
            "FTC",
            "GAPFAC",
            "GSA",
            "HHS",
            "HUD",
            "NARA",
            "NASA",
            "NCUA",
            "NLRB",
            "NRC",
            "NSF",
            "NTSB",
            "ONCD",
            "OPM",
            "PBGC",
            "PCLOB",
            "SBA",
            "SSA",
            "TREAS",
            "USC",
            "USDA",
            "VA",
            "ACF",
            "ACL",
            "AHRQ",
            "AID",
            "AMS",
            "AOA",
            "APHIS",
            "ARS",
            "ASC",
            "ATBCB",
            "ATF",
            "ATR",
            "ATSDR",
            "BIA",
            "BIS",
            "BLM",
            "BLS",
            "BOEM",
            "BOP",
            "BOR",
            "BPA",
            "BPD",
            "BSC",
            "BSEE",
            "CCC",
            "CDC",
            "CDFI",
            "CEQ",
            "CFPB",
            "CISA",
            "CMS",
            "CNCS",
            "COE",
            "COLC",
            "CPPBSD",
            "CPSC",
            "CSB",
            "CSEO",
            "CSREES",
            "DARS",
            "DEA",
            "DEPO",
            "DHS",
            "DOC",
            "DOD",
            "DOE",
            "DOI",
            "DOJ",
            "DOL",
            "DOS",
            "DOT",
            "EAB",
            "EAC",
            "EBSA",
            "ECAB",
            "ECSA",
            "ED",
            "EDA",
            "EEOC",
            "EERE",
            "EIA",
            "EIB",
            "EOA",
            "EOIR",
            "EPA",
            "ERS",
            "ESA",
            "ETA",
            "FAA",
            "FAR",
            "FAS",
            "FBI",
            "FCIC",
            "FCSC",
            "FDA",
            "FEMA",
            "FFIEC",
            "FHWA",
            "FINCEN",
            "FIRSTNET",
            "FISCAL",
            "FLETC",
            "FMC",
            "FMCSA",
            "FNS",
            "FPAC",
            "FRA",
            "FRTIB",
            "FS",
            "FSA",
            "FSIS",
            "FSOC",
            "FTA",
            "FTC",
            "FTZB",
            "FWS",
            "GAPFAC",
            "GIPSA",
            "GSA",
            "HHS",
            "HHSIG",
            "HRSA",
            "HUD",
            "IACB",
            "ICEB",
            "IHS",
            "IPEC",
            "IRS",
            "ISOO",
            "ITA",
            "LMSO",
            "MARAD",
            "MBDA",
            "MMS",
            "MSHA",
            "NAL",
            "NARA",
            "NASA",
            "NASS",
            "NCS",
            "NCUA",
            "NEO",
            "NHTSA",
            "NIC",
            "NIFA",
            "NIGC",
            "NIH",
            "NIST",
            "NLRB",
            "NNSA",
            "NOAA",
            "NPS",
            "NRC",
            "NRCS",
            "NSF",
            "NSPC",
            "NTIA",
            "NTSB",
            "OCC",
            "OEPNU",
            "OFAC",
            "OFCCP",
            "OFPP",
            "OFR",
            "OJJDP",
            "OJP",
            "OMB",
            "ONCD",
            "ONDCP",
            "ONRR",
            "OPM",
            "OPPM",
            "OSHA",
            "OSM",
            "OSTP",
            "OTS",
            "PBGC",
            "PCLOB",
            "PHMSA",
            "PTO",
            "RBS",
            "RHS",
            "RITA",
            "RMA",
            "RTB",
            "RUS",
            "SAMHSA",
            "SBA",
            "SEPA",
            "SLSDC",
            "SSA",
            "SWPA",
            "TA",
            "TREAS",
            "TSA",
            "TTB",
            "USA",
            "USAF",
            "USBC",
            "USCBP",
            "USCG",
            "USCIS",
            "USDA",
            "USDAIG",
            "USGS",
            "USMINT",
            "USN",
            "USPC",
            "USTR",
            "VA",
            "VETS",
            "WAPA",
            "WCPO",
            "WHD",
        ]

    def fetch_all_pages(self, endpoint, params, max_items_per_page=250):
        """
        Iterates through all the pages of API data for a given endpoint
        until there are no more pages to fetch (occurs at 20 pages, or
        5000 items).

        Args:
            endpoint (str): The API endpoint to fetch data from.
                            ['dockets', 'documents', 'comments']
            params (dict): Dictionary of parameters to send in the API request.
            max_items_per_page (int): Maximum number of items per page to
            request. Max (default) = 250.

        Returns:
            list: A list of items (dictionaries) fetched from all pages of the
                API endpoint.
        """
        items = []
        page = 1
        while True:
            params["page[number]"] = page
            params["page[size]"] = max_items_per_page

            response = requests.get(
                f"{self.base_url}/{endpoint}",
                headers=self.headers,
                params=params,
            )
            print(f"Requesting: {response.url}")

            if response.status_code == 200:
                try:
                    data = response.json()
                except ValueError:
                    print("Failed to decode JSON response")
                    break
                items.extend(data["data"])
                if not data["meta"].get("hasNextPage", False):
                    break
                page += 1
            elif response.status_code == 429:  # Rate limit exceeded
                retry_after = response.headers.get(
                    "Retry-After", None
                )  # Obtain reset time
                wait_time = (
                    int(retry_after)
                    if retry_after and retry_after.isdigit()
                    else 3600
                )  # Default to 1 hour if no wait time provided
                print(
                    f"""Rate limit exceeded.
                    Waiting to {wait_time} seconds to retry."""
                )
                time.sleep(wait_time)
                continue
            else:
                print(f"Error fetching page {page}: {response.status_code}")
                break
        return items

    def get_all_dockets_by_agency(self):
        """
        Retrieves all docket IDs by looping through predefined agencies and
        stores them in a CSV file.
        """
        all_dockets = []

        for agency in self.agencies:
            filter_params = {"filter[agencyId]": agency}

            agency_dockets = self.fetch_all_pages("dockets", filter_params)

            for docket in agency_dockets:
                # Extract relevant information from each docket
                docket_details = {
                    "docket_id": docket.get("id"),
                    "docket_type": docket.get("attributes", {}).get(
                        "docketType"
                    ),
                    "last_modified": docket.get("attributes", {}).get(
                        "lastModifiedDate"
                    ),
                    "agency_id": docket.get("attributes", {}).get("agencyId"),
                    "title": docket.get("attributes", {}).get("title"),
                    "obj_id": docket.get("attributes", {}).get("objectId"),
                }
                all_dockets.append(docket_details)

        # Store as pandas dataframe.
        # We can change this mode of storage if you have a different idea,
        # I think this is a sensible approach, to limit use of our API calls.
        df = pd.DataFrame(all_dockets)
        df.drop_duplicates(
            subset=["docket_id"], inplace=True
        )  # Ensure there are no duplicate dockets based on docket_id
        df.to_csv("dockets_detailed.csv", index=False)

    def fetch_documents_by_date_ranges(self, start_date, end_date):
        """
        Fetches documents posted within specified date ranges and saves them
        to a CSV file.

        Args:
            start_date (datetime.date): Start date for fetching documents.
            end_date (datetime.date): End date for fetching documents.
        """
        all_documents = []
        for start, end in self.generate_date_ranges(start_date, end_date):
            print(f"Fetching documents from {start} to {end}")
            params = {
                "filter[postedDate][ge]": start.strftime("%Y-%m-%d"),
                "filter[postedDate][le]": end.strftime("%Y-%m-%d"),
            }
            documents = self.fetch_all_pages("documents", params)
            all_documents.extend(documents)

        print(f"Total documents fetched: {len(all_documents)}")

        # Extract relevant data from documents
        document_lst = []
        for document in all_documents:
            doc_data = {
                "Doc_ID": document.get("id"),
                "Doc_Type": document.get("attributes", {}).get("documentType"),
                "Last_Modified": document.get("attributes", {}).get(
                    "lastModifiedDate"
                ),
                "FR_Doc_Num": document.get("attributes", {}).get("frDocNum"),
                "Withdrawn": document.get("attributes", {}).get("withdrawn"),
                "Agency_ID": document.get("attributes", {}).get("agencyId"),
                "Comment_End_Date": document.get("attributes", {}).get(
                    "commentEndDate"
                ),
                "Title": document.get("attributes", {}).get("title"),
                "Posted_Date": document.get("attributes", {}).get("postedDate"),
                "Docket_ID": document.get("attributes", {}).get("docketId"),
                "Subtype": document.get("attributes", {}).get("subtype"),
                "Comment_Start_Date": document.get("attributes", {}).get(
                    "commentStartDate"
                ),
                "Open_For_Comment": document.get("attributes", {}).get(
                    "openForComment"
                ),
                "Object_ID": document.get("attributes", {}).get("objectId"),
            }
            document_lst.append(doc_data)

        # Save to DataFrame and CSV
        df = pd.DataFrame(document_lst)
        df = df.drop_duplicates()
        df.to_csv("doc_detailed_2024.csv", index=False)

    @staticmethod  # for now, we can put this in utils if that is preferred.
    def generate_date_ranges(start_date, end_date):
        """
        Generates weekly date ranges between two dates, inclusive.
        Helped function for fetch_documents_by_date_ranges().

        Args:
            start_date (datetime.date): The start date of the range.
            end_date (datetime.date): The end date of the range.

        Yields:
            tuple: A tuple of (start_date, end_date) for each week within the
                specified range.
        """
        current_date = start_date
        while current_date < end_date:
            week_end = current_date + datetime.timedelta(days=6)
            yield (current_date, min(week_end, end_date))
            current_date = week_end + datetime.timedelta(days=1)

    def fetch_comment_count_by_documents(self, document_ids, file_output_path):
        """
        Fetches comments count for each document ID that is open for comments.

        Args:
            document_ids (DataFrame): DataFrame containing document IDs under
            the column 'Object_ID'.
            This can be obtained from the output of
            fetch_documents_by_date_ranges()
            file_output_path (str): Path to save the output csv file.

        Returns:
            None: Results are saved directly to a csv file specified by
                file_output_path.
        """
        base_url = f"{self.base_url}/comments"
        results = []

        for commentId in document_ids["Object_ID"]:
            continue_fetching = True
            while continue_fetching:
                params = {"filter[commentOnId]": commentId}

                response = requests.get(
                    base_url, headers=self.headers, params=params
                )
                if response.status_code == 200:
                    data = response.json()
                    total_elements = data["meta"]["totalElements"]
                    results.append(
                        {"id": commentId, "total_elements": total_elements}
                    )
                    continue_fetching = False
                elif response.status_code == 429:  # Rate limit exceeded
                    retry_after = response.headers.get("Retry-After", None)
                    wait_time = (
                        int(retry_after)
                        if retry_after and retry_after.isdigit()
                        else 3600
                    )
                    print(
                        f"""Rate limit exceeded.
                        Waiting {wait_time} seconds to retry."""
                    )
                    time.sleep(wait_time)
                else:
                    results.append(
                        {"id": commentId, "total_elements": "Failed to fetch"}
                    )
                    continue_fetching = False

        results_df = pd.DataFrame(results)
        results_df.to_csv(file_output_path)

__init__(api_key)

Initializes the BulkDl class.

Parameters:

Name	Type	Description	Default
api_key	str	API key for authenticating requests to the	required

Attributes:

Name	Type	Description
api_key	str	Stored API key for requests.
base_url	str	Base URL for the API endpoint.
headers	dict	Headers to include in API requests, containing API key and content type.
agencies	list[str]	List of agency identifiers (aggregated from https://www.regulations.gov/agencies) to be used in API calls.

Source code in civiclens/collect/bulk_dl.py

def __init__(self, api_key):
    """
    Initializes the BulkDl class.

    Args:
        api_key (str): API key for authenticating requests to the
        regulations.gov API.

    Attributes:
        api_key (str): Stored API key for requests.
        base_url (str): Base URL for the API endpoint.
        headers (dict): Headers to include in API requests, containing API
            key and content type.
        agencies (list[str]): List of agency identifiers (aggregated from
            https://www.regulations.gov/agencies) to be used in API calls.
    """
    self.api_key = api_key
    self.base_url = "https://api.regulations.gov/v4"
    self.headers = {
        "X-Api-Key": self.api_key,
        "Content-Type": "application/json",
    }
    self.agencies = [
        "AID",
        "ATBCB",
        "CFPB",
        "CNCS",
        "COLC",
        "CPPBSD",
        "CPSC",
        "CSB",
        "DHS",
        "DOC",
        "DOD",
        "DOE",
        "DOI",
        "DOJ",
        "DOL",
        "DOS",
        "DOT",
        "ED",
        "EEOC",
        "EIB",
        "EOP",
        "EPA",
        "FFIEC",
        "FMC",
        "FRTIB",
        "FTC",
        "GAPFAC",
        "GSA",
        "HHS",
        "HUD",
        "NARA",
        "NASA",
        "NCUA",
        "NLRB",
        "NRC",
        "NSF",
        "NTSB",
        "ONCD",
        "OPM",
        "PBGC",
        "PCLOB",
        "SBA",
        "SSA",
        "TREAS",
        "USC",
        "USDA",
        "VA",
        "ACF",
        "ACL",
        "AHRQ",
        "AID",
        "AMS",
        "AOA",
        "APHIS",
        "ARS",
        "ASC",
        "ATBCB",
        "ATF",
        "ATR",
        "ATSDR",
        "BIA",
        "BIS",
        "BLM",
        "BLS",
        "BOEM",
        "BOP",
        "BOR",
        "BPA",
        "BPD",
        "BSC",
        "BSEE",
        "CCC",
        "CDC",
        "CDFI",
        "CEQ",
        "CFPB",
        "CISA",
        "CMS",
        "CNCS",
        "COE",
        "COLC",
        "CPPBSD",
        "CPSC",
        "CSB",
        "CSEO",
        "CSREES",
        "DARS",
        "DEA",
        "DEPO",
        "DHS",
        "DOC",
        "DOD",
        "DOE",
        "DOI",
        "DOJ",
        "DOL",
        "DOS",
        "DOT",
        "EAB",
        "EAC",
        "EBSA",
        "ECAB",
        "ECSA",
        "ED",
        "EDA",
        "EEOC",
        "EERE",
        "EIA",
        "EIB",
        "EOA",
        "EOIR",
        "EPA",
        "ERS",
        "ESA",
        "ETA",
        "FAA",
        "FAR",
        "FAS",
        "FBI",
        "FCIC",
        "FCSC",
        "FDA",
        "FEMA",
        "FFIEC",
        "FHWA",
        "FINCEN",
        "FIRSTNET",
        "FISCAL",
        "FLETC",
        "FMC",
        "FMCSA",
        "FNS",
        "FPAC",
        "FRA",
        "FRTIB",
        "FS",
        "FSA",
        "FSIS",
        "FSOC",
        "FTA",
        "FTC",
        "FTZB",
        "FWS",
        "GAPFAC",
        "GIPSA",
        "GSA",
        "HHS",
        "HHSIG",
        "HRSA",
        "HUD",
        "IACB",
        "ICEB",
        "IHS",
        "IPEC",
        "IRS",
        "ISOO",
        "ITA",
        "LMSO",
        "MARAD",
        "MBDA",
        "MMS",
        "MSHA",
        "NAL",
        "NARA",
        "NASA",
        "NASS",
        "NCS",
        "NCUA",
        "NEO",
        "NHTSA",
        "NIC",
        "NIFA",
        "NIGC",
        "NIH",
        "NIST",
        "NLRB",
        "NNSA",
        "NOAA",
        "NPS",
        "NRC",
        "NRCS",
        "NSF",
        "NSPC",
        "NTIA",
        "NTSB",
        "OCC",
        "OEPNU",
        "OFAC",
        "OFCCP",
        "OFPP",
        "OFR",
        "OJJDP",
        "OJP",
        "OMB",
        "ONCD",
        "ONDCP",
        "ONRR",
        "OPM",
        "OPPM",
        "OSHA",
        "OSM",
        "OSTP",
        "OTS",
        "PBGC",
        "PCLOB",
        "PHMSA",
        "PTO",
        "RBS",
        "RHS",
        "RITA",
        "RMA",
        "RTB",
        "RUS",
        "SAMHSA",
        "SBA",
        "SEPA",
        "SLSDC",
        "SSA",
        "SWPA",
        "TA",
        "TREAS",
        "TSA",
        "TTB",
        "USA",
        "USAF",
        "USBC",
        "USCBP",
        "USCG",
        "USCIS",
        "USDA",
        "USDAIG",
        "USGS",
        "USMINT",
        "USN",
        "USPC",
        "USTR",
        "VA",
        "VETS",
        "WAPA",
        "WCPO",
        "WHD",
    ]

fetch_all_pages(endpoint, params, max_items_per_page=250)

Iterates through all the pages of API data for a given endpoint until there are no more pages to fetch (occurs at 20 pages, or 5000 items).

Parameters:

Name	Type	Description	Default
endpoint	str	The API endpoint to fetch data from. ['dockets', 'documents', 'comments']	required
params	dict	Dictionary of parameters to send in the API request.	required
max_items_per_page	int	Maximum number of items per page to	250

Returns:

Name	Type	Description
list		A list of items (dictionaries) fetched from all pages of the API endpoint.

Source code in civiclens/collect/bulk_dl.py

def fetch_all_pages(self, endpoint, params, max_items_per_page=250):
    """
    Iterates through all the pages of API data for a given endpoint
    until there are no more pages to fetch (occurs at 20 pages, or
    5000 items).

    Args:
        endpoint (str): The API endpoint to fetch data from.
                        ['dockets', 'documents', 'comments']
        params (dict): Dictionary of parameters to send in the API request.
        max_items_per_page (int): Maximum number of items per page to
        request. Max (default) = 250.

    Returns:
        list: A list of items (dictionaries) fetched from all pages of the
            API endpoint.
    """
    items = []
    page = 1
    while True:
        params["page[number]"] = page
        params["page[size]"] = max_items_per_page

        response = requests.get(
            f"{self.base_url}/{endpoint}",
            headers=self.headers,
            params=params,
        )
        print(f"Requesting: {response.url}")

        if response.status_code == 200:
            try:
                data = response.json()
            except ValueError:
                print("Failed to decode JSON response")
                break
            items.extend(data["data"])
            if not data["meta"].get("hasNextPage", False):
                break
            page += 1
        elif response.status_code == 429:  # Rate limit exceeded
            retry_after = response.headers.get(
                "Retry-After", None
            )  # Obtain reset time
            wait_time = (
                int(retry_after)
                if retry_after and retry_after.isdigit()
                else 3600
            )  # Default to 1 hour if no wait time provided
            print(
                f"""Rate limit exceeded.
                Waiting to {wait_time} seconds to retry."""
            )
            time.sleep(wait_time)
            continue
        else:
            print(f"Error fetching page {page}: {response.status_code}")
            break
    return items

fetch_comment_count_by_documents(document_ids, file_output_path)

Fetches comments count for each document ID that is open for comments.

Parameters:

Name	Type	Description	Default
document_ids	DataFrame	DataFrame containing document IDs under	required
file_output_path	str	Path to save the output csv file.	required

Returns:

Name	Type	Description
None		Results are saved directly to a csv file specified by file_output_path.

Source code in civiclens/collect/bulk_dl.py

def fetch_comment_count_by_documents(self, document_ids, file_output_path):
    """
    Fetches comments count for each document ID that is open for comments.

    Args:
        document_ids (DataFrame): DataFrame containing document IDs under
        the column 'Object_ID'.
        This can be obtained from the output of
        fetch_documents_by_date_ranges()
        file_output_path (str): Path to save the output csv file.

    Returns:
        None: Results are saved directly to a csv file specified by
            file_output_path.
    """
    base_url = f"{self.base_url}/comments"
    results = []

    for commentId in document_ids["Object_ID"]:
        continue_fetching = True
        while continue_fetching:
            params = {"filter[commentOnId]": commentId}

            response = requests.get(
                base_url, headers=self.headers, params=params
            )
            if response.status_code == 200:
                data = response.json()
                total_elements = data["meta"]["totalElements"]
                results.append(
                    {"id": commentId, "total_elements": total_elements}
                )
                continue_fetching = False
            elif response.status_code == 429:  # Rate limit exceeded
                retry_after = response.headers.get("Retry-After", None)
                wait_time = (
                    int(retry_after)
                    if retry_after and retry_after.isdigit()
                    else 3600
                )
                print(
                    f"""Rate limit exceeded.
                    Waiting {wait_time} seconds to retry."""
                )
                time.sleep(wait_time)
            else:
                results.append(
                    {"id": commentId, "total_elements": "Failed to fetch"}
                )
                continue_fetching = False

    results_df = pd.DataFrame(results)
    results_df.to_csv(file_output_path)

fetch_documents_by_date_ranges(start_date, end_date)

Fetches documents posted within specified date ranges and saves them to a CSV file.

Parameters:

Name	Type	Description	Default
start_date	date	Start date for fetching documents.	required
end_date	date	End date for fetching documents.	required

Source code in civiclens/collect/bulk_dl.py

def fetch_documents_by_date_ranges(self, start_date, end_date):
    """
    Fetches documents posted within specified date ranges and saves them
    to a CSV file.

    Args:
        start_date (datetime.date): Start date for fetching documents.
        end_date (datetime.date): End date for fetching documents.
    """
    all_documents = []
    for start, end in self.generate_date_ranges(start_date, end_date):
        print(f"Fetching documents from {start} to {end}")
        params = {
            "filter[postedDate][ge]": start.strftime("%Y-%m-%d"),
            "filter[postedDate][le]": end.strftime("%Y-%m-%d"),
        }
        documents = self.fetch_all_pages("documents", params)
        all_documents.extend(documents)

    print(f"Total documents fetched: {len(all_documents)}")

    # Extract relevant data from documents
    document_lst = []
    for document in all_documents:
        doc_data = {
            "Doc_ID": document.get("id"),
            "Doc_Type": document.get("attributes", {}).get("documentType"),
            "Last_Modified": document.get("attributes", {}).get(
                "lastModifiedDate"
            ),
            "FR_Doc_Num": document.get("attributes", {}).get("frDocNum"),
            "Withdrawn": document.get("attributes", {}).get("withdrawn"),
            "Agency_ID": document.get("attributes", {}).get("agencyId"),
            "Comment_End_Date": document.get("attributes", {}).get(
                "commentEndDate"
            ),
            "Title": document.get("attributes", {}).get("title"),
            "Posted_Date": document.get("attributes", {}).get("postedDate"),
            "Docket_ID": document.get("attributes", {}).get("docketId"),
            "Subtype": document.get("attributes", {}).get("subtype"),
            "Comment_Start_Date": document.get("attributes", {}).get(
                "commentStartDate"
            ),
            "Open_For_Comment": document.get("attributes", {}).get(
                "openForComment"
            ),
            "Object_ID": document.get("attributes", {}).get("objectId"),
        }
        document_lst.append(doc_data)

    # Save to DataFrame and CSV
    df = pd.DataFrame(document_lst)
    df = df.drop_duplicates()
    df.to_csv("doc_detailed_2024.csv", index=False)

generate_date_ranges(start_date, end_date) staticmethod

Generates weekly date ranges between two dates, inclusive. Helped function for fetch_documents_by_date_ranges().

Parameters:

Name	Type	Description	Default
start_date	date	The start date of the range.	required
end_date	date	The end date of the range.	required

Yields:

Name	Type	Description
tuple		A tuple of (start_date, end_date) for each week within the specified range.

Source code in civiclens/collect/bulk_dl.py

@staticmethod  # for now, we can put this in utils if that is preferred.
def generate_date_ranges(start_date, end_date):
    """
    Generates weekly date ranges between two dates, inclusive.
    Helped function for fetch_documents_by_date_ranges().

    Args:
        start_date (datetime.date): The start date of the range.
        end_date (datetime.date): The end date of the range.

    Yields:
        tuple: A tuple of (start_date, end_date) for each week within the
            specified range.
    """
    current_date = start_date
    while current_date < end_date:
        week_end = current_date + datetime.timedelta(days=6)
        yield (current_date, min(week_end, end_date))
        current_date = week_end + datetime.timedelta(days=1)

get_all_dockets_by_agency()

Retrieves all docket IDs by looping through predefined agencies and stores them in a CSV file.

Source code in civiclens/collect/bulk_dl.py

def get_all_dockets_by_agency(self):
    """
    Retrieves all docket IDs by looping through predefined agencies and
    stores them in a CSV file.
    """
    all_dockets = []

    for agency in self.agencies:
        filter_params = {"filter[agencyId]": agency}

        agency_dockets = self.fetch_all_pages("dockets", filter_params)

        for docket in agency_dockets:
            # Extract relevant information from each docket
            docket_details = {
                "docket_id": docket.get("id"),
                "docket_type": docket.get("attributes", {}).get(
                    "docketType"
                ),
                "last_modified": docket.get("attributes", {}).get(
                    "lastModifiedDate"
                ),
                "agency_id": docket.get("attributes", {}).get("agencyId"),
                "title": docket.get("attributes", {}).get("title"),
                "obj_id": docket.get("attributes", {}).get("objectId"),
            }
            all_dockets.append(docket_details)

    # Store as pandas dataframe.
    # We can change this mode of storage if you have a different idea,
    # I think this is a sensible approach, to limit use of our API calls.
    df = pd.DataFrame(all_dockets)
    df.drop_duplicates(
        subset=["docket_id"], inplace=True
    )  # Ensure there are no duplicate dockets based on docket_id
    df.to_csv("dockets_detailed.csv", index=False)