caseydm

import grequests

def test_rate_limit_no_key():
    # test url 20 times at once
    # 10 of the responses are 429, rate limited
    urls = []
    for i in range(20):
        urls.append(
            "https://api.openalex.org/works?filter=doi:10.1016/j.jfca.2023.105165")
    rs = (grequests.get(u) for u in urls)

caseydm

import requests


def main():
    result = {}
    doi_count = 0
    no_landing_page_in_s3 = 0
    for page in range(1, 11):
        url = f"https://api.openalex.org/works?page={page}&per-page=100&sample=1000&seed=23&filter=has_doi:true"
        r1 = requests.get(url)

caseydm

import csv

from app import db

with open("publishers_image_thumbnail_urls.csv", "r") as f:
    reader = csv.reader(f)
    for row in reader:
        publisher_id = int(row[0].replace("https://openalex.org/P", ""))
        image_thumbnail_url = row[2]
        db.session.execute('UPDATE mid.publisher SET image_thumbnail_url = :image_thumbnail_url WHERE publisher_id = :publisher_id', {"image_thumbnail_url": image_thumbnail_url, "publisher_id": publisher_id})

caseydm

import csv
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import time


def safe_nested_get(dct, keys, default=None):
    """
    Safely get a nested value from a dictionary.