88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
import logging
|
|
import secrets
|
|
|
|
import requests
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from profiles.models import UserProfile
|
|
|
|
from vrobbler.settings import TODOIST_CLIENT_ID
|
|
|
|
logger = logging.getLogger(__name__)
|
|
User = get_user_model()
|
|
|
|
TODOIST_OAUTH_START_URL = "https://todoist.com/oauth/authorize?client_id={id}&scope=data:read_write&state=".format(
|
|
id=TODOIST_CLIENT_ID
|
|
)
|
|
TODOIST_OAUTH_TOKEN_URL = "https://todoist.com/oauth/access_token"
|
|
TODOIST_API_URL = "https://api.todoist.com/api/v1/sync"
|
|
|
|
|
|
def generate_todoist_oauth_url(user_id: int) -> str:
|
|
user_profile = User.objects.filter(id=user_id).first().profile
|
|
user_profile.todoist_state = secrets.token_hex(16)
|
|
user_profile.save(update_fields=["todoist_state"])
|
|
return TODOIST_OAUTH_START_URL + user_profile.todoist_state
|
|
|
|
|
|
def get_todoist_access_token(user_id: int, state: str, code: str):
|
|
logger.info(
|
|
"[get_todoist_access_token] called",
|
|
extra={
|
|
"user_id": user_id,
|
|
"state": state,
|
|
"state_repr": repr(state),
|
|
"code": code,
|
|
},
|
|
)
|
|
|
|
user_profile = UserProfile.objects.filter(user_id=user_id).first()
|
|
|
|
logger.info(
|
|
"[get_todoist_access_token] found profile",
|
|
extra={
|
|
"user_id": user_id,
|
|
"profile_state": user_profile.todoist_state if user_profile else None,
|
|
"passed_state": state,
|
|
"match": user_profile.todoist_state == state if user_profile else False,
|
|
},
|
|
)
|
|
|
|
if not user_profile or user_profile.todoist_state != state:
|
|
logger.error("[get_todoist_access_token] profile not found or state mismatch")
|
|
raise Exception("Could not find profile or state mismatch")
|
|
|
|
post_data = {
|
|
"client_id": settings.TODOIST_CLIENT_ID,
|
|
"client_secret": settings.TODOIST_CLIENT_SECRET,
|
|
"code": code,
|
|
}
|
|
|
|
response = requests.post(TODOIST_OAUTH_TOKEN_URL, data=post_data)
|
|
|
|
if response.status_code == 200:
|
|
access_token = response.json().get("access_token")
|
|
user_profile.todoist_auth_key = access_token
|
|
user_profile.todoist_state = None
|
|
user_profile.save(update_fields=["todoist_auth_key", "todoist_state"])
|
|
|
|
sync_response = requests.post(
|
|
TODOIST_API_URL,
|
|
data={"sync_token": "*", "resource_types": '["user"]'},
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
if sync_response.status_code == 200:
|
|
todoist_user_id = sync_response.json().get("user", {}).get("id")
|
|
if todoist_user_id:
|
|
user_profile.todoist_user_id = todoist_user_id
|
|
user_profile.save(update_fields=["todoist_user_id"])
|
|
logger.info(
|
|
"[get_todoist_access_token] set todoist_user_id",
|
|
extra={"todoist_user_id": todoist_user_id},
|
|
)
|
|
|
|
logger.info(
|
|
"[get_todoist_access_token] finished",
|
|
extra={"user_id": user_id},
|
|
)
|