Skip to content

Merge and simplify session/token logic #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
timo-piirmann wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Merge and simplify session/token logic #2

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 37 additions & 35 deletions index.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,36 @@

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
session = requests_cache.CachedSession(expire_after=60*15)
load_dotenv()
token = None


def get_token() -> str:
logger.info(f"Getting token for {os.environ.get('PLUTOF_USER')}")
url = "https://api.plutof.ut.ee/v1/public/auth/token/"
headers = {"User-Agent": "eDNA sample tracker"}
client_data = {
"username": os.environ.get("PLUTOF_USER"),
"password": os.environ.get("PLUTOF_PASSWORD"),
"client_id": os.environ.get("PLUTOF_CLIENT_ID"),
"client_secret": os.environ.get("PLUTOF_CLIENT_SECRET"),
"grant_type": "password",
}
res = requests.post(url, data=client_data, headers=headers)
data = res.json()

return data["access_token"]


token = get_token()
session = requests_cache.CachedSession(expire_after=60*15)
session.headers.update({
"User-Agent": "eDNA sample tracker",
"Authorization": f"Bearer {token}",
})


@dataclass
class Site():
class Site:
name: str | None = None
simplified_name: str | None = None
url: str | None = None
Expand All @@ -30,13 +53,13 @@ class Site():


@dataclass
class Dna():
class Dna:
plutof_id: int | None = None
concentration: float | None = None


@dataclass
class Sample():
class Sample:
plutof_id: int | None = None
blank: bool | None = False
name: str | None = None
Expand All @@ -57,37 +80,18 @@ class Sample():
dnas: list[Dna] = field(default_factory=list)


def get_token() -> str:
global token
if token is None:
logger.info(f"Getting token for {os.environ.get('PLUTOF_USER')}")
url = "https://api.plutof.ut.ee/v1/public/auth/token/"
res = requests.post(url, data={
"username": os.environ.get("PLUTOF_USER"),
"password": os.environ.get("PLUTOF_PASSWORD"),
"client_id": os.environ.get("PLUTOF_CLIENT_ID"),
"client_secret": os.environ.get("PLUTOF_CLIENT_SECRET"),
"grant_type": "password"
})
data = res.json()
token = data["access_token"]
return token


def paginate(url: str, use_data: bool=False) -> list:
token = get_token()
def paginate(url: str, use_data: bool = False) -> list:
page = 1
items = []

while True:
page_url = url + str(page)
logger.debug(page_url)
res = session.get(page_url, headers={
"Authorization": f"Bearer {token}",
"User-Agent": "eDNA sample tracker"
})
res = session.get(page_url)

if res.status_code == 403:
raise Exception("Forbidden 403")
if res.status_code != 200:
elif res.status_code != 200:
break
else:
results = res.json()
Expand All @@ -103,12 +107,8 @@ def paginate(url: str, use_data: bool=False) -> list:


def get_object(url: str) -> dict | None:
token = get_token()
logger.debug(url)
res = session.get(url, headers={
"Authorization": f"Bearer {token}",
"User-Agent": "eDNA sample tracker"
})
res = session.get(url)
if res.status_code != 200:
return None
else:
Expand Down Expand Up @@ -144,7 +144,7 @@ def fetch_dnas_for_samples(samples: list[dict]) -> dict[int, dict]:
dna_dict = dict()
sample_ids = set((sample["id"] for sample in samples))
for sample_id in sample_ids:
url = f"https://api.plutof.ut.ee/v1/dna-lab/dnas/?include=dna_extraction&material_sample={sample_id}&ordering=-id&page[size]=20&page[number]="
url = f"https://api.plutof.ut.ee/v1/dna-lab/dnas/?include=dna_extraction&material_sample={sample_id}&ordering=-id&page[size]=20&page[number]="
dnas = paginate(url, use_data=True)
if len(dnas) > 0:
dna_dict[sample_id] = dnas
Expand Down Expand Up @@ -179,12 +179,14 @@ def find_id(url: str) -> int:
def simplify_name(name: str) -> str:
return re.sub(r"_+", "_", re.sub(r"\W+", "_", unidecode(name).lower()))


def get_site_info_dict():
site_dict = dict()
for site in site_info:
site_dict[simplify_name(site["name"])] = site
return site_dict


def main():

result_samples = []
Expand Down