e.vlasov/redmine_tracker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

повысил надежность обработки API-ответов и ошибок БД

Browse Source
This commit is contained in:
Власов Эдуард
2026-07-06 16:07:08 +03:00
parent bd20ef4d48
commit 64b347e966
5 changed files with 76 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
api/clockify.py
View File

@@ -33,12 +33,16 @@ class ClockifyAPI(ClockifyConfig):
page-size: int - Размер страницы (по умолчанию 50)
"""
time_entries_url = self.time_entries_url
activities = requests.get(
resp = requests.get(
self.base_url + time_entries_url,
params=kwargs,
headers={"X-API-KEY": self.token},
timeout=REQUEST_TIMEOUT,
).json()
)
if not resp.ok:
logger.error(f"Ошибка ClockifyAPI: {resp.text} {resp.status_code}")
resp.raise_for_status()
activities = resp.json()
if not activities:
return []
logger.info(f"Получено задач: {len(activities)}")
@@ -84,6 +88,9 @@ class ClockifyAPI(ClockifyConfig):
task_id = parse_task_id(tag.title)
author = db.get_user_by("clockify_user_id", activity["userId"])
if not author:
logger.error(f"Не найден автор активности {activity['id']}")
continue
result.append(
Activity(
@@ -124,12 +131,16 @@ class ClockifyAPI(ClockifyConfig):
**kwargs,
"page": page
}
tags = requests.get(
resp = requests.get(
self.base_url + self.tags_url,
params=params,
headers={"X-API-KEY": self.token},
timeout=REQUEST_TIMEOUT,
).json()
)
if not resp.ok:
logger.error(f"Ошибка ClockifyAPI: {resp.text} {resp.status_code}")
resp.raise_for_status()
tags = resp.json()
for tag in tags:
clockify_tag = Tag(
@@ -156,18 +167,26 @@ class ClockifyAPI(ClockifyConfig):
"""
Получает с сервера Clockify Проекты (ДО) Ветро
"""
projects = requests.get(
resp = requests.get(
self.base_url + self.projects_url,
headers={"X-API-KEY": self.token},
timeout=REQUEST_TIMEOUT,
).json()
)
if not resp.ok:
logger.error(f"Ошибка ClockifyAPI: {resp.text} {resp.status_code}")
resp.raise_for_status()
projects = resp.json()
for project in projects:
if not project.get("archived"):
vetro_projects = requests.get(
resp = requests.get(
self.base_url + self.tasks_url(project.get("id")),
headers={"X-API-KEY": self.token},
timeout=REQUEST_TIMEOUT,
).json()
)
if not resp.ok:
logger.error(f"Ошибка ClockifyAPI: {resp.text} {resp.status_code}")
resp.raise_for_status()
vetro_projects = resp.json()
return [VetroProject(
id=vetro_project["id"],
title=vetro_project["name"]
@@ -200,7 +219,7 @@ class ClockifyAPI(ClockifyConfig):
if config.is_debug or resp.ok:
logger.info(f"{Fore.GREEN}{activity}{Fore.RESET}".ljust(60) + "✅")
else:
logger.debug(resp.json())
logger.debug(resp.text)
logger.error(f"{Fore.RED}{activity}{Fore.RESET}".ljust(60) + "❌")
except Exception as e:
logger.error(e)
@@ -214,7 +233,7 @@ class ClockifyAPI(ClockifyConfig):
timeout=REQUEST_TIMEOUT,
)
if not resp.ok:
logger.error(resp.json())
logger.error(resp.text)
raise InvalidToken
resp = resp.json()
logger.debug({"id": resp["id"], "email": resp["email"], "name": resp["name"]})
@@ -228,7 +247,7 @@ class ClockifyAPI(ClockifyConfig):
timeout=REQUEST_TIMEOUT,
)
if not resp.ok:
logger.error(resp.json())
logger.error(resp.text)
raise requests.exceptions.RequestException(Fore.RED + "Ошибка запроса." + Fore.RESET)
json = resp.json()
return {

39
api/redmine.py
View File

@@ -80,20 +80,23 @@ class RedmineAPI(RedmineConfig):
logger.info(f"{Fore.RED}{activity}{Fore.RESET}".ljust(60) + "❌")
return tracked
def get_time_entry_activities(self) -> list[ActivityTypes]:
def get_time_entry_activities(self) -> list[ActivityType]:
"""Получает виды Деятельности с сервера Redmine"""
resp = requests.get(
self.base_url + self.time_entry_activities,
headers={"X-Redmine-API-KEY": self.token},
timeout=REQUEST_TIMEOUT,
)
logger.debug(resp.json())
if resp.ok:
return [
ActivityType(id=act.get("id"), title=act.get("name"))
for act in resp.json().get("time_entry_activities")
if act.get("active")
]
if not resp.ok:
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
return []
data = resp.json()
logger.debug(data)
return [
ActivityType(id=act.get("id"), title=act.get("name"))
for act in data.get("time_entry_activities", [])
if act.get("active")
]
def get_user(self):
"""Проверяет токен и возвращает данные юзера"""
@@ -103,7 +106,7 @@ class RedmineAPI(RedmineConfig):
timeout=REQUEST_TIMEOUT,
)
if not resp.ok:
logger.error("Invalid Token for RedmineAPI")
logger.error(f"Invalid Token for RedmineAPI: {resp.text}")
raise InvalidToken
json = resp.json()["user"]
return {
@@ -114,7 +117,7 @@ class RedmineAPI(RedmineConfig):
"username": json["login"]
}
def get_issue(self, issue_id: int) -> Task:
def get_issue(self, issue_id: int) -> Task | None:
"""Получает задачу по её id с сервера Redmine"""
resp = requests.get(
self.base_url + self.issue.format(**{"id": issue_id}),
@@ -122,8 +125,8 @@ class RedmineAPI(RedmineConfig):
timeout=REQUEST_TIMEOUT,
)
if not resp.ok:
logger.error("Ошибка RedmineAPI")
return {}
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
return None
issue = resp.json()["issue"]
return Task(**{
"id": issue["id"],
@@ -146,10 +149,14 @@ class RedmineAPI(RedmineConfig):
params={"limit": 100, "offset": 100 * i, "status_id": "*"},
timeout=REQUEST_TIMEOUT,
)
if not resp.ok or not resp.json():
logger.error("Ошибка RedmineAPI")
return {}
issues.extend(resp.json()["issues"])
if not resp.ok:
logger.error(f"Ошибка RedmineAPI: {resp.text} {resp.status_code}")
return []
data = resp.json()
if not data:
logger.error("Ошибка RedmineAPI: пустой ответ")
return []
issues.extend(data["issues"])
return [Task(**{
"id": issue["id"],
"subject": issue["subject"],

13
db/constants.py Normal file
View File

@@ -0,0 +1,13 @@
"""Константы слоя базы данных."""
USER_LOOKUP_FIELDS = frozenset(
{
"id",
"clockify_user_id",
"redmine_user_id",
"clockify_email",
"clockify_username",
"redmine_email",
"redmine_username",
}
)

11
db/db.py
View File

@@ -5,6 +5,7 @@ from colorama import Fore
from classes import Activity, ActivityType, Tag, VetroProject
from classes.classes import Task
from db.constants import USER_LOOKUP_FIELDS
from enums import ActivityTypes
logger = logging.getLogger(__name__)
@@ -102,8 +103,9 @@ class Database:
)
)
self.con.commit()
except sq.IntegrityError:
pass
except sq.IntegrityError as e:
self.con.rollback()
logger.error(f"Не удалось сохранить трудочасы по активности {act.id}: {e}")
def insert_tags(self, tags: list[Tag]):
@@ -238,8 +240,13 @@ class Database:
def get_user_by(self, by_field: str, value: str):
"""Получает пользователя из бд"""
if by_field not in USER_LOOKUP_FIELDS:
raise ValueError(f"Недопустимое поле поиска пользователя: {by_field}")
try:
user = self.cursor.execute(f"SELECT * FROM user WHERE {by_field} = ?", (value,)).fetchone()
if not user:
logger.error(f"Пользователь не найден: {by_field}={value}")
return None
return {
"id": user[0],
"first_name": user[1],

2
utils.py
View File

@@ -25,7 +25,7 @@ def today_end() -> datetime:
year = td.year
month = td.month
day = td.day
return datetime(year=year, month=month, day=day, hour=23)
return datetime(year=year, month=month, day=day, hour=23, minute=59, second=59)
def str_to_date(date: str) -> datetime | None: