Работа с mymsuapi

[Zimovchik]

Изменения

Детали реализации

Check-List

• [x] Вы проверили свой код перед отправкой запроса?
• [ ] Вы написали тесты к реализованным функциям?
• [x] Вы не забыли применить форматирование black и isort для Back-End или Prettier для Front-End?

  55

[github-actions[bot]]

:poop: Code linting failed, use black and isort to fix it.

[github-actions[bot]]

Summary

Tests	Skipped	Failures	Errors	Time
1	0 :zzz:	0 :x:	0 :fire:	3.903s :stopwatch:

[Zimovchik]

Функция которая забирает данные из api

import json
from datetime import datetime, timedelta
import logging
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

def get_group_schedule(group_number: int | str, course_id: int, flow_id: int, date_from: datetime, date_to: datetime):
    url = "https://api.test.my.msu.ru/gateway/public/api/v1/public_content/lessons"
    values_to_keep = ["id", "date", "time_from", "time_to", "schedule_id", "discipline", "classroom", "conducting_way", "lesson_type", "teacher_users", "study_groups"]
    params = {
        "schedule_id[]": "1",
        "course_id": str(course_id),
        "flow_id": str(flow_id),
        "date_from": date_from.strftime("%Y-%m-%d"),
        "date_to": date_to.strftime("%Y-%m-%d")
    }
    try:
        response = requests.get(url, params=params)
        logging.info(f"Successfully fetched schedule for group {group_number} from {date_from} to {date_to}. status: {response.status_code}")
        if response.status_code == 200:
            lessons = response.json()["result"]["data"]
        else:
            return None
    except requests.exceptions.RequestException as e:
        logging.error(f"Error fetching schedule for group {group_number}: {e}")
        return None
    lessons = filter(lambda lesson: (
                    (lesson["study_groups"][0]["type"] == "simple" and lesson["study_groups"][0]["name"] == str(group_number)))
                    or (lesson["study_groups"][0]["type"] == "separation" and lesson["study_groups"][0]["base_groups"][0]["name"] == str(group_number))
                    or ((lesson["study_groups"][0]["type"] == "union" and any(group["name"] == str(group_number) for group in lesson["study_groups"][0]["base_groups"]))),
                    lessons)
    lessons = list(lessons)
    res = []
    for lesson in lessons:
        res.append({key : lesson[key] for key in values_to_keep})
    return res

def insert_schedule_into_db(schedule_data: list[dict]):
    """Загружает данные в stg таблицу RawTimetableApi

    Args:
        :param schedule_data: принимает данные, которые возвращает функция get_group_schedule
    """

[Zimovchik]

55

Написать функцию insert_schedule_into_db. (Видимо надо сразу делать dag)

[Zimovchik]

@mixx3 Я посмотрел все ручки в сваггере которые были закрыты (((