Comprehensive Source Code for a Telegram Mini Web Application: Includes Backend, Frontend, and In-Depth Documentation
from .base import Base
from .diagnostics import Diagnostic, DiagnosticLocation
from .doctors import Doctor, DoctorRating
from .locations import Location
from .slots import Booking
from .users import User

backend/src/medsyncapp/infrastructure/database/models/__init__.py
Latand/MedSyncWebApp
from datetime import datetime

from sqlalchemy.dialects.postgresql import TIMESTAMP
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.sql.functions import func
from typing_extensions import Annotated

int_pk = Annotated[int, mapped_column(primary_key=True)]


class Base(DeclarativeBase):
    pass


class TableNameMixin:
    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower() + "s"


class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(TIMESTAMP, server_default=func.now())

backend/src/medsyncapp/infrastructure/database/models/base.py
Latand/MedSyncWebApp
from typing import Optional

from sqlalchemy import String, ForeignKey, DECIMAL
from sqlalchemy.orm import Mapped, mapped_column

from .base import Base, TableNameMixin, int_pk, TimestampMixin


class Diagnostic(Base, TableNameMixin):
    diagnostic_id: Mapped[int_pk]
    type_name: Mapped[str] = mapped_column(String(128))
    description: Mapped[str] = mapped_column(String(256))
    price: Mapped[float] = mapped_column(DECIMAL(10, 2))
    photo_url: Mapped[str] = mapped_column(String(256))



class DiagnosticLocation(Base):
    __tablename__ = "diagnostic_locations"
    diagnostic_location_id: Mapped[int_pk]
    diagnostic_id: Mapped[int] = mapped_column(ForeignKey("diagnostics.diagnostic_id"))
    location_id: Mapped[int] = mapped_column(ForeignKey("locations.location_id"))


class DiagnosticResult(Base, TimestampMixin):
    __tablename__ = "diagnostic_results"
    diagnostic_result_id: Mapped[int_pk]
    booking_id = mapped_column(ForeignKey("bookings.booking_id"))
    file_path: Mapped[str] = mapped_column(String(256))
    file_id: Mapped[Optional[str]] = mapped_column(String(256))
backend/src/medsyncapp/infrastructure/database/models/diagnostics.py
Latand/MedSyncWebApp
from typing import Optional

from sqlalchemy import Integer, String, ForeignKey, Enum, DECIMAL, TEXT
from sqlalchemy.orm import Mapped, mapped_column

from .base import Base, TableNameMixin, TimestampMixin, int_pk


class Doctor(Base, TableNameMixin):
    doctor_id: Mapped[int_pk]
    location_id: Mapped[int] = mapped_column(ForeignKey("locations.location_id"))
    full_name: Mapped[str] = mapped_column(String(128))
    specialty_id: Mapped[int] = mapped_column(ForeignKey("specialties.specialty_id"))
    price: Mapped[float] = mapped_column(DECIMAL(10, 2))
    photo_url: Mapped[str] = mapped_column(String(256))
    experience: Mapped[Optional[str]] = mapped_column(TEXT)
    certificates: Mapped[Optional[str]] = mapped_column(TEXT)
    services: Mapped[Optional[str]] = mapped_column(TEXT)


class Specialty(Base, TableNameMixin):
    __tablename__ = "specialties"
    specialty_id: Mapped[int_pk]
    specialty_name: Mapped[str] = mapped_column(String(128))


class DoctorRating(Base):
    __tablename__ = "doctor_ratings"
    rating_id: Mapped[int_pk]
    doctor_id: Mapped[int] = mapped_column(ForeignKey("doctors.doctor_id"))
    rating: Mapped[int] = mapped_column(Integer, default=0)


backend/src/medsyncapp/infrastructure/database/models/doctors.py
Latand/MedSyncWebApp
from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column

from .base import Base, TableNameMixin, int_pk


class Location(Base, TableNameMixin):
    location_id: Mapped[int_pk]
    name: Mapped[String] = mapped_column(String(128))
    address: Mapped[String] = mapped_column(String(256))
backend/src/medsyncapp/infrastructure/database/models/locations.py
Latand/MedSyncWebApp
from datetime import datetime
from typing import Optional

from sqlalchemy import (
    String,
    ForeignKey,
    Boolean,
    false,
    Integer,
    TIMESTAMP,
    Date,
    UniqueConstraint,
)
from sqlalchemy.orm import Mapped, mapped_column

from .base import Base, TimestampMixin, TableNameMixin, int_pk


class WorkingHour(Base, TimestampMixin, TableNameMixin):
    working_hour_id: Mapped[int_pk]
    location_id: Mapped[int] = mapped_column(ForeignKey("locations.location_id"))
    start_time: Mapped[int] = mapped_column(Integer)
    end_time: Mapped[int] = mapped_column(Integer)
    weekday_index: Mapped[int] = mapped_column(Integer)


class Booking(Base, TimestampMixin, TableNameMixin):
    booking_id: Mapped[int_pk]
    user_id: Mapped[Optional[int]] = mapped_column(ForeignKey("users.user_id"))
    user_full_name: Mapped[str] = mapped_column(String(256))
    user_email: Mapped[str] = mapped_column(String(256))
    user_phone_number: Mapped[str] = mapped_column(String(16))
    user_message: Mapped[Optional[str]] = mapped_column(String(1024))

    doctor_id: Mapped[Optional[int]] = mapped_column(ForeignKey("doctors.doctor_id", ondelete="SET NULL"))
    diagnostic_id: Mapped[Optional[int]] = mapped_column(ForeignKey("diagnostics.diagnostic_id", ondelete="SET NULL"))
    location_id: Mapped[int] = mapped_column(ForeignKey("locations.location_id", ondelete="SET NULL"))
    booking_time: Mapped[datetime] = mapped_column(TIMESTAMP(timezone=True))

    __table_args__ = (
        UniqueConstraint('doctor_id', 'location_id', 'booking_time', name='unique_doctor_location_time'),
        UniqueConstraint('diagnostic_id', 'location_id', 'booking_time', name='unique_diagnostic_location_time'),
    )
backend/src/medsyncapp/infrastructure/database/models/slots.py
Latand/MedSyncWebApp
from typing import Optional

from sqlalchemy import String
from sqlalchemy import text, BIGINT, Boolean, true
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

from .base import Base, TimestampMixin, TableNameMixin


class User(Base, TimestampMixin, TableNameMixin):
    """
    This class represents a User in the application.
    If you want to learn more about SQLAlchemy and Alembic, you can check out the following link to my course:
    https://www.udemy.com/course/sqlalchemy-alembic-bootcamp/?referralCode=E9099C5B5109EB747126

    Attributes:
        user_id (Mapped[int]): The unique identifier of the user.
        username (Mapped[Optional[str]]): The username of the user.
        full_name (Mapped[str]): The full name of the user.

    Methods:
        __repr__(): Returns a string representation of the User object.

    Inherited Attributes:
        Inherits from Base, TimestampMixin, and TableNameMixin classes, which provide additional attributes and functionality.

    Inherited Methods:
        Inherits methods from Base, TimestampMixin, and TableNameMixin classes, which provide additional functionality.

    """
    user_id: Mapped[int] = mapped_column(BIGINT, primary_key=True, autoincrement=False)
    username: Mapped[Optional[str]] = mapped_column(String(128))
    full_name: Mapped[str] = mapped_column(String(128))

    def __repr__(self):
        return f"<User {self.user_id} {self.username} {self.full_name}>"
backend/src/medsyncapp/infrastructure/database/models/users.py
Latand/MedSyncWebApp
from sqlalchemy.ext.asyncio import AsyncSession


class BaseRepo:
    """
    A class representing a base repository for handling database operations.

    Attributes:
        session (AsyncSession): The database session used by the repository.

    """

    def __init__(self, session):
        self.session: AsyncSession = session

backend/src/medsyncapp/infrastructure/database/repo/base.py
Latand/MedSyncWebApp
from sqlalchemy import select

from medsyncapp.infrastructure.database.models import Booking, Doctor, Diagnostic, Location
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class BookingsRepo(BaseRepo):
    async def get_booking(self, booking_id: int):
        query = (
            select(Booking, Doctor, Diagnostic, Location)
            .outerjoin(Doctor, Doctor.doctor_id == Booking.doctor_id)
            .outerjoin(Diagnostic, Diagnostic.diagnostic_id == Booking.diagnostic_id)
            .join(Location, Location.location_id == Booking.location_id)
            .where(Booking.booking_id == booking_id)
        )
        return (await self.session.execute(query)).first()


    async def get_user_bookings(self, user_id: int):
        query = (
            select(Booking, Doctor, Diagnostic, Location)
            .outerjoin(Doctor, Doctor.doctor_id == Booking.doctor_id)
            .outerjoin(Diagnostic, Diagnostic.diagnostic_id == Booking.diagnostic_id)
            .join(Location, Location.location_id == Booking.location_id)
            .where(Booking.user_id == user_id)
            .order_by(Booking.booking_time.desc())
        )
        return (await self.session.execute(query)).all()
backend/src/medsyncapp/infrastructure/database/repo/bookings.py
Latand/MedSyncWebApp
import datetime

from dateutil.parser import parse
from sqlalchemy import func
from sqlalchemy import select, insert

from medsyncapp.infrastructure.database.models import (
    Diagnostic,
    Booking,
    DiagnosticLocation,
    Location,
)
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class DiagnosticRepo(BaseRepo):
    async def get_all_diagnostic_types(self):
        stmt = (
            select(
                Diagnostic.diagnostic_id,
                Diagnostic.type_name,
                Diagnostic.description,
                Diagnostic.price,
                Diagnostic.photo_url,
                func.count(DiagnosticLocation.diagnostic_id).label("clinics_count"),
            )
            .join(
                DiagnosticLocation,
                Diagnostic.diagnostic_id == DiagnosticLocation.diagnostic_id,
            )
            .group_by(Diagnostic.diagnostic_id)
            .order_by(Diagnostic.diagnostic_id)
        )
        result = await self.session.execute(stmt)
        return result.mappings().all()

    async def get_locations_by_type(self, diagnostic_id: int):
        stmt = (
            select(Location.location_id, Location.name, Location.address)
            .join(
                DiagnosticLocation,
                Location.location_id == DiagnosticLocation.location_id,
            )
            .where(DiagnosticLocation.diagnostic_id == diagnostic_id)
        )
        result = await self.session.execute(stmt)
        return result.mappings().all()

    async def book_slot(
        self,
        payload: dict,
        user_id: int = None,
    ):
        # Insert into Booking
        insert_stmt = (
            insert(Booking)
            .values(
                user_id=user_id,
                user_full_name=payload.get("user_name", "")
                + " "
                + payload.get("user_surname", ""),
                user_email=payload.get("user_email"),
                user_phone_number=payload.get("user_phone"),
                user_message=payload.get("user_message"),
                diagnostic_id=payload.get("diagnostic_id"),
                location_id=payload.get("location_id"),
                booking_time=parse(payload.get("booking_date_time")),
            )
            .returning(Booking.booking_id)
        )
        result = await self.session.scalar(insert_stmt)
        await self.session.commit()

        return result

    async def get_booked_slots(
        self, diagnostic_id: int, location_id: int, month_number: int
    ) -> list[Booking]:
        stmt = select(Booking.booking_time).where(
            Booking.diagnostic_id == diagnostic_id,
            Booking.booking_time >= datetime.date.today(),
            func.extract("month", Booking.booking_time) == month_number + 1,
            Booking.location_id == location_id,
        )
        result = await self.session.scalars(stmt)
        return result.all()

backend/src/medsyncapp/infrastructure/database/repo/diagnostics.py
Latand/MedSyncWebApp
import datetime

from dateutil.parser import parse
from sqlalchemy import select, insert, func

from medsyncapp.infrastructure.database.models import (
    Doctor,
    Booking,
    DoctorRating,
    Location,
)
from medsyncapp.infrastructure.database.models.doctors import Specialty
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class DoctorRepo(BaseRepo):
    async def get_all_doctors(self) -> list[Doctor]:
        stmt = (
            select(
                Doctor.doctor_id,
                Doctor.location_id,
                Doctor.full_name,
                Specialty.specialty_name,
                Specialty.specialty_id,
                Doctor.price,
                Doctor.photo_url,
                Location.name.label("location_name"),
                Location.address.label("location_address"),
                func.coalesce(func.avg(DoctorRating.rating), 0).label("avg_rating"),
                func.coalesce(func.count(DoctorRating.rating_id), 0).label("reviews"),
            )
            .join(Location, Location.location_id == Doctor.location_id)
            .join(Specialty, Specialty.specialty_id == Doctor.specialty_id)
            .outerjoin(DoctorRating, DoctorRating.doctor_id == Doctor.doctor_id)
            .group_by(Doctor.doctor_id, Location.location_id, Specialty.specialty_id)
        )

        result = await self.session.execute(stmt)
        result = result.mappings().all()
        return result

    async def get_specialties(self) -> list[Specialty]:
        stmt = (
            select(Specialty.specialty_id, Specialty.specialty_name)
            .join(Doctor)
            .order_by(Specialty.specialty_name)
            .group_by(Specialty.specialty_id)
        )
        result = await self.session.execute(stmt)
        return result.mappings().all()

    async def book_slot(self, payload: dict, user_id=None):
        insert_stmt = (
            insert(Booking)
            .values(
                user_id=user_id,
                user_full_name=payload.get("user_name", "")
                + " "
                + payload.get("user_surname", ""),
                user_email=payload.get("user_email"),
                user_phone_number=payload.get("user_phone"),
                user_message=payload.get("user_message"),
                doctor_id=payload.get("doctor_id"),
                location_id=payload.get("location_id"),
                booking_time=parse(payload.get("booking_date_time")),
            )
            .returning(Booking.booking_id)
        )
        result = await self.session.scalar(insert_stmt)
        await self.session.commit()
        return result

    async def get_doctor(self, doctor_id: int) -> Doctor:
        stmt = (
            select(
                Doctor.doctor_id,
                Doctor.location_id,
                Doctor.full_name,
                Specialty.specialty_name,
                Specialty.specialty_id,
                Doctor.price,
                Doctor.photo_url,
                Location.address.label("location_address"),
                Doctor.experience,
                Doctor.certificates,
                Location.name.label("location_name"),
                func.coalesce(func.avg(DoctorRating.rating), 0).label("avg_rating"),
                func.coalesce(func.count(DoctorRating.rating_id), 0).label("reviews"),
                Doctor.services,
            )
            .join(Location, Location.location_id == Doctor.location_id)
            .outerjoin(DoctorRating, DoctorRating.doctor_id == Doctor.doctor_id)
            .join(Specialty, Specialty.specialty_id == Doctor.specialty_id)
            .where(Doctor.doctor_id == doctor_id)
            .group_by(Doctor.doctor_id, Location.location_id, Specialty.specialty_id)
        )

        result = await self.session.execute(stmt)
        return result.mappings().first()

    async def get_booked_slots(
        self, doctor_id: int, location_id: int, month_number: int
    ) -> list[Booking]:
        stmt = select(Booking.booking_time).where(
            Booking.doctor_id == doctor_id,
            Booking.booking_time >= datetime.date.today(),
            func.extract("month", Booking.booking_time) == month_number + 1,
            Booking.location_id == location_id,
        )
        result = await self.session.scalars(stmt)
        return result.all()

backend/src/medsyncapp/infrastructure/database/repo/doctors.py
Latand/MedSyncWebApp
from sqlalchemy import select

from medsyncapp.infrastructure.database.models.locations import Location
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class LocationsRepo(BaseRepo):
    async def get_location(self, location_id: int) -> Location:
        query = select(
            Location.location_id,
            Location.name,
            Location.address,
        ).where(Location.location_id == location_id)
        return (await self.session.execute(query)).mappings().first()

backend/src/medsyncapp/infrastructure/database/repo/locations.py
Latand/MedSyncWebApp
from dataclasses import dataclass

from sqlalchemy.ext.asyncio import AsyncSession

from medsyncapp.infrastructure.database.repo.bookings import BookingsRepo
from medsyncapp.infrastructure.database.repo.diagnostics import DiagnosticRepo
from medsyncapp.infrastructure.database.repo.doctors import DoctorRepo
from medsyncapp.infrastructure.database.repo.locations import LocationsRepo
from medsyncapp.infrastructure.database.repo.results import ResultsRepo
from medsyncapp.infrastructure.database.repo.slots import SlotRepo
from medsyncapp.infrastructure.database.repo.users import UserRepo


@dataclass
class RequestsRepo:
    """
    Repository for handling database operations. This class holds all the repositories for the database models.

    You can add more repositories as properties to this class, so they will be easily accessible.
    """

    session: AsyncSession

    @property
    def users(self) -> UserRepo:
        return UserRepo(self.session)

    @property
    def doctors(self) -> DoctorRepo:
        return DoctorRepo(self.session)

    @property
    def diagnostics(self) -> DiagnosticRepo:
        return DiagnosticRepo(self.session)

    @property
    def slots(self) -> SlotRepo:
        return SlotRepo(self.session)

    @property
    def locations(self) -> LocationsRepo:
        return LocationsRepo(self.session)

    @property
    def bookings(self) -> BookingsRepo:
        return BookingsRepo(self.session)

    @property
    def results(self) -> ResultsRepo:
        return ResultsRepo(self.session)

backend/src/medsyncapp/infrastructure/database/repo/requests.py
Latand/MedSyncWebApp
from sqlalchemy import select, update
from sqlalchemy.dialects.postgresql import insert

from medsyncapp.infrastructure.database.models import Booking
from medsyncapp.infrastructure.database.models.diagnostics import DiagnosticResult, Diagnostic
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class ResultsRepo(BaseRepo):
    async def get_results(self, user_id: int):
        query = (
            select(DiagnosticResult, Booking, Diagnostic)
            .join(Booking, Booking.booking_id == DiagnosticResult.booking_id)
            .join(Diagnostic, Diagnostic.diagnostic_id == Booking.diagnostic_id)
            .where(Booking.user_id == user_id)
            .order_by(Booking.booking_time.desc())
        )
        results = await self.session.execute(query)
        return results.all()

    async def get_result(self, result_id: str):
        query = (
            select(DiagnosticResult, Booking, Diagnostic)
            .join(Booking, Booking.booking_id == DiagnosticResult.booking_id)
            .join(Diagnostic, Diagnostic.diagnostic_id == Booking.diagnostic_id)
            .where(DiagnosticResult.diagnostic_result_id == result_id)
        )
        result = await self.session.execute(query)
        return result.first()

    async def save_file_id(self, result_id: str, file_id: str):
        query = (
            update(DiagnosticResult)
            .where(DiagnosticResult.diagnostic_result_id == result_id)
            .values(file_id=file_id)
        )
        await self.session.execute(query)
        await self.session.commit()

    async def create_result(self, booking_id: int, diagnostic_id: int):
        default_paths_mapping = {
            1: "results/mri.pdf",
            2: "results/ct.pdf",
            3: "results/x-ray.pdf",
            4: "results/ultrasound.pdf",
            5: "results/pet.pdf",
            6: "results/blood-test.pdf",
            7: "results/urine.pdf",
            8: "results/biopsy.pdf",
            9: "results/ecg.pdf",
            10: "results/bone-density.pdf",
            11: "results/endoscopy.pdf",
            12: "results/colonoscopy.pdf",
        }

        await self.session.execute(
            insert(DiagnosticResult).values(
                booking_id=booking_id,
                file_path=default_paths_mapping[diagnostic_id],
            )
        )
        await self.session.commit()

backend/src/medsyncapp/infrastructure/database/repo/results.py
Latand/MedSyncWebApp
from sqlalchemy import select

from medsyncapp.infrastructure.database.models.slots import WorkingHour
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class SlotRepo(BaseRepo):
    async def get_working_hours(self, location_id: int):
        query = select(WorkingHour).where(WorkingHour.location_id == location_id)
        return (await self.session.scalars(query)).all()

backend/src/medsyncapp/infrastructure/database/repo/slots.py
Latand/MedSyncWebApp
from typing import Optional

from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert

from medsyncapp.infrastructure.database.models import User
from medsyncapp.infrastructure.database.repo.base import BaseRepo


class UserRepo(BaseRepo):
    async def get_or_create_user(
        self,
        user_id: int,
        full_name: str,
        username: Optional[str] = None,
    ):
        """
        Creates or updates a new user in the database and returns the user object.
        :param user_id: The user's ID.
        :param full_name: The user's full name.
        :param username: The user's username. It's an optional parameter.
        :return: User object, None if there was an error while making a transaction.
        """

        insert_stmt = (
            insert(User)
            .values(
                user_id=user_id,
                username=username,
                full_name=full_name,
            )
            .on_conflict_do_update(
                index_elements=[User.user_id],
                set_=dict(
                    username=username,
                    full_name=full_name,
                ),
            )
            .returning(User)
        )
        result = await self.session.execute(insert_stmt)

        await self.session.commit()
        return result.scalar_one()

backend/src/medsyncapp/infrastructure/database/repo/users.py
Latand/MedSyncWebApp
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

from medsyncapp.tgbot.config import DbConfig


def create_engine(db: DbConfig, echo=False):
    engine = create_async_engine(
        db.construct_sqlalchemy_url(),
        query_cache_size=1200,
        pool_size=20,
        max_overflow=200,
        future=True,
        echo=echo,
    )
    return engine


def create_session_pool(engine):
    session_pool = async_sessionmaker(bind=engine, expire_on_commit=False)
    return session_pool

backend/src/medsyncapp/infrastructure/database/setup.py
Latand/MedSyncWebApp
import asyncio
from logging.config import fileConfig

from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config

from alembic import context

from medsyncapp.tgbot.config import load_config

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config

# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

env_config = load_config(".env")
url = env_config.db.construct_sqlalchemy_url()

config.set_main_option(
    "sqlalchemy.url",
    url
)

# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
from medsyncapp.infrastructure.database.models import Base
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode.

    This configures the context with just a URL
    and not an Engine, though an Engine is acceptable
    here as well.  By skipping the Engine creation
    we don't even need a DBAPI to be available.

    Calls to context.execute() here emit the given string to the
    script output.

    """
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """In this scenario we need to create an Engine
    and associate a connection with the context.

    """

    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""

    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

backend/src/medsyncapp/migrations/env.py
Latand/MedSyncWebApp
"""add booking constraint

Revision ID: 0b7dc85ebb76
Revises: 566ce5e78aa3
Create Date: 2023-10-01 18:27:08.099205

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '0b7dc85ebb76'
down_revision: Union[str, None] = '566ce5e78aa3'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_unique_constraint('unique_diagnostic_location_time', 'bookings', ['diagnostic_id', 'location_id', 'booking_time'])
    op.create_unique_constraint('unique_doctor_location_time', 'bookings', ['doctor_id', 'location_id', 'booking_time'])
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint('unique_doctor_location_time', 'bookings', type_='unique')
    op.drop_constraint('unique_diagnostic_location_time', 'bookings', type_='unique')
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/0b7dc85ebb76_add_booking_constraint.py
Latand/MedSyncWebApp
"""add photo to diagnostics

Revision ID: 0bfc6022b536
Revises: 54b5265d2e43
Create Date: 2023-10-02 19:00:03.256724

"""
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision: str = "0bfc6022b536"
down_revision: Union[str, None] = "54b5265d2e43"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

diagnostics_photo_mapping = {
    1: "/images/get-tested/mri.png",
    2: "/images/get-tested/ct.png",
    3: "/images/get-tested/x-ray.png",
    4: "/images/get-tested/ultrasound.png",
    5: "/images/get-tested/pet.png",
    6: "/images/get-tested/blood-test.png",
    7: "/images/get-tested/urine.png",
    8: "/images/get-tested/biopsy.png",
    9: "/images/get-tested/ecg.png",
    10: "/images/get-tested/bone-density.png",
    11: "/images/get-tested/endoscopy.png",
    12: "/images/get-tested/colonoscopy.png",
}


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column("diagnostics", sa.Column("photo_url", sa.String(length=256)))
    conn = op.get_bind()

    for diagnostic_id, photo_url in diagnostics_photo_mapping.items():
        conn.execute(
            sa.text(
                "UPDATE diagnostics SET photo_url = :photo_url WHERE diagnostic_id = :diagnostic_id"
            ),
            parameters=dict(
                diagnostic_id=diagnostic_id,
                photo_url=photo_url,
            ),
        )

    op.alter_column("diagnostics", "photo_url", nullable=False)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column("diagnostics", "photo_url")
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/0bfc6022b536_add_photo_to_diagnostics.py
Latand/MedSyncWebApp
"""seed new data

Revision ID: 0e5cf4b909ff
Revises: 7b0bae0ba599
Create Date: 2023-09-29 11:58:40.668065

"""
import random
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op
from faker import Faker

# revision identifiers, used by Alembic.
revision: str = '0e5cf4b909ff'
down_revision: Union[str, None] = '7b0bae0ba599'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

male_photos = [1, 2, 3, 5, 6, 10]
female_photos = [4, 7, 8, 9, 11, 12]

diagnostics_list = [
    {
        "type_name": "Magnetic Resonance Imaging",
        "description": "A procedure using magnetic fields and radio waves to create detailed images of the organs and tissues within the body.",
    },
    {
        "type_name": "Computed Tomography",
        "description": "An imaging procedure using X-rays and computer technology to produce cross-sectional images of the body.",
    },
    {
        "type_name": "X-Ray",
        "description": "A test using small amounts of radiation to produce images of the structures inside the body, especially bones.",
    },
    {
        "type_name": "Ultrasound",
        "description": "An imaging technique using high-frequency sound waves to create images of structures within the body.",
    },
    {
        "type_name": "Positron Emission Tomography",
        "description": "A procedure that helps show how tissues and organs are functioning, using a radioactive drug to show activity.",
    },
    {
        "type_name": "Blood Tests",
        "description": "Tests performed on a blood sample to check the function of certain organs and to assess health.",
    },
    {
        "type_name": "Urine Tests",
        "description": "Tests performed on a urine sample to check for various disorders, including those related to the kidneys and urinary tract.",
    },
    {
        "type_name": "Biopsy",
        "description": "A procedure where a piece of tissue or a sample of cells is removed from the body to be examined under a microscope.",
    },
    {
        "type_name": "Electrocardiogram",
        "description": "A test that measures the electrical activity of the heartbeat to identify abnormalities.",
    },
    {
        "type_name": "Bone Density Test",
        "description": "A test that measures bone mineral content to check for bone loss and osteoporosis.",
    },
    {
        "type_name": "Endoscopy",
        "description": "A procedure to examine the digestive tract using a flexible tube with a light and camera attached to it.",
    },
    {
        "type_name": "Colonoscopy",
        "description": "A procedure to examine the inside of the colon and rectum, using a long, flexible tube with a camera on the end.",
    },
]

doctor_specialties = [
    "Cardiologist",
    "Dermatologist",
    "Endocrinologist",
    "Gastroenterologist",
    "Gynecologist",
    "Hematologist",
    "Neurologist",
    "Ophthalmologist",
    "Otolaryngologist",
]


def upgrade() -> None:
    # Get the database connection
    conn = op.get_bind()
    # Initialize Faker
    fake = Faker()

    # Seed Locations
    location_ids = []
    for _ in range(3):
        result = conn.execute(
            sa.text(
                "INSERT INTO locations (name, address) VALUES (:name, :address) RETURNING location_id;",
            ),
            parameters=dict(
                name=fake.company(),
                address=fake.address()[:25],
            ),
        )
        location_id = result.scalar()
        location_ids.append(location_id)

    # Seed Diagnostics
    diagnostic_ids = []
    for diagnostic_type in diagnostics_list:
        result = conn.execute(
            sa.text(
                "INSERT INTO diagnostics (type_name, description, price) VALUES (:type_name, :description, :price) RETURNING diagnostic_id;",
            ),
            parameters=dict(
                type_name=diagnostic_type["type_name"],
                description=diagnostic_type["description"],
                price=float("{:.2f}".format(random.uniform(50.0, 500.0))),
            ),
        )
        diagnostic_id = result.scalar()
        diagnostic_ids.append(diagnostic_id)

    # Seed Diagnostic_Locations
    for diag_id in diagnostic_ids:
        for loc_id in random.sample(location_ids, 3):
            conn.execute(
                sa.text(
                    "INSERT INTO diagnostic_locations (diagnostic_id, location_id) VALUES (:diagnostic_id, :location_id);",
                ),
                parameters=dict(
                    diagnostic_id=diag_id,
                    location_id=loc_id,
                ),
            )
    speciality_ids = []
    # Seed Specialties
    for title in doctor_specialties:
        result = conn.scalar(
            sa.text(
                "INSERT INTO specialties (specialty_name) VALUES (:title) RETURNING specialty_id;",
            ),
            parameters=dict(
                title=title,
            ),
        )
        speciality_ids.append(result)

    # Seed Doctors
    for gender, photo_ids in [("male", male_photos), ("female", female_photos)]:
        for photo_id in photo_ids:
            conn.execute(
                sa.text(
                    "INSERT INTO doctors (location_id, full_name, specialty_id, price, photo_url) VALUES (:location_id, :full_name, :specialty_id, :price, :photo_url) RETURNING doctor_id;"
                ),
                parameters=dict(
                    location_id=random.choice(location_ids),
                    full_name=fake.name_male()
                    if gender == "male"
                    else fake.name_female(),
                    specialty_id=random.choice(speciality_ids),
                    price=float("{:.2f}".format(random.uniform(50.0, 200.0))),
                    photo_url=f"/images/doctors-listing/profiles/{photo_id}.png",
                ),
            )


def downgrade() -> None:
    pass

backend/src/medsyncapp/migrations/versions/0e5cf4b909ff_seed_new_data.py
Latand/MedSyncWebApp
"""removed slots completely

Revision ID: 2651025c84a1
Revises: a6cf7fd5d39b
Create Date: 2023-09-30 20:51:19.674759

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

# revision identifiers, used by Alembic.
revision: str = '2651025c84a1'
down_revision: Union[str, None] = 'a6cf7fd5d39b'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ##
    op.drop_constraint('bookings_diagnostic_slot_id_fkey', 'bookings', type_='foreignkey')
    op.drop_constraint('bookings_doctor_slot_id_fkey', 'bookings', type_='foreignkey')#
    op.execute(sa.text("DROP TABLE slots CASCADE"))
    op.drop_table('doctor_slots')
    op.drop_table('diagnostic_slots')
    op.drop_table('doctor_bookings')
    op.add_column('bookings', sa.Column('user_id', sa.BIGINT(), nullable=False))
    op.add_column('bookings', sa.Column('user_message', sa.String(length=1024), nullable=True))
    op.add_column('bookings', sa.Column('doctor_id', sa.Integer(), nullable=True))
    op.add_column('bookings', sa.Column('diagnostic_id', sa.Integer(), nullable=True))
    op.add_column('bookings', sa.Column('location_id', sa.Integer(), nullable=False))
    op.add_column('bookings', sa.Column('booking_time', sa.TIMESTAMP(), nullable=False))
    op.create_foreign_key(None, 'bookings', 'locations', ['location_id'], ['location_id'], ondelete='SET NULL')
    op.create_foreign_key(None, 'bookings', 'diagnostics', ['diagnostic_id'], ['diagnostic_id'], ondelete='SET NULL')
    op.create_foreign_key(None, 'bookings', 'doctors', ['doctor_id'], ['doctor_id'], ondelete='SET NULL')
    op.create_foreign_key(None, 'bookings', 'users', ['user_id'], ['user_id'])
    op.drop_column('bookings', 'doctor_slot_id')
    op.drop_column('bookings', 'diagnostic_slot_id')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('bookings', sa.Column('diagnostic_slot_id', sa.INTEGER(), autoincrement=False, nullable=True))
    op.add_column('bookings', sa.Column('doctor_slot_id', sa.INTEGER(), autoincrement=False, nullable=True))
    # op.drop_constraint(None, 'bookings', type_='foreignkey')
    # op.drop_constraint(None, 'bookings', type_='foreignkey')
    # op.drop_constraint(None, 'bookings', type_='foreignkey')
    # op.drop_constraint(None, 'bookings', type_='foreignkey')
    op.drop_column('bookings', 'booking_time')
    op.drop_column('bookings', 'location_id')
    op.drop_column('bookings', 'diagnostic_id')
    op.drop_column('bookings', 'doctor_id')
    op.drop_column('bookings', 'user_message')
    op.drop_column('bookings', 'user_id')
    op.create_table('slots',
    sa.Column('slot_id', sa.INTEGER(), server_default=sa.text("nextval('slots_slot_id_seq'::regclass)"), autoincrement=True, nullable=False),
    sa.Column('location_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('start_time', postgresql.TIMESTAMP(), autoincrement=False, nullable=False),
    sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), autoincrement=False, nullable=False),
    sa.Column('date', sa.DATE(), autoincrement=False, nullable=False),
    sa.Column('is_booked', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
    sa.ForeignKeyConstraint(['location_id'], ['locations.location_id'], name='slots_location_id_fkey'),
    sa.PrimaryKeyConstraint('slot_id', name='slots_pkey'),
    postgresql_ignore_search_path=False
    )
    op.create_table('diagnostic_slots',
    sa.Column('diagnostic_slot_id', sa.INTEGER(), autoincrement=True, nullable=False),
    sa.Column('diagnostic_location_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('slot_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.ForeignKeyConstraint(['diagnostic_location_id'], ['diagnostic_locations.diagnostic_location_id'], name='diagnostic_slots_diagnostic_location_id_fkey', ondelete='CASCADE'),
    sa.ForeignKeyConstraint(['slot_id'], ['slots.slot_id'], name='diagnostic_slots_slot_id_fkey', ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('diagnostic_slot_id', name='diagnostic_slots_pkey')
    )
    op.create_table('doctor_bookings',
    sa.Column('appointment_id', sa.INTEGER(), autoincrement=True, nullable=False),
    sa.Column('user_id', sa.BIGINT(), autoincrement=False, nullable=False),
    sa.Column('doctor_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('slot_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('status', postgresql.ENUM('Booked', 'Cancelled', name='booking_status', create_type=False), autoincrement=False, nullable=False),
    sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), autoincrement=False, nullable=False),
    sa.ForeignKeyConstraint(['doctor_id'], ['doctors.doctor_id'], name='doctor_bookings_doctor_id_fkey'),
    sa.ForeignKeyConstraint(['slot_id'], ['slots.slot_id'], name='doctor_bookings_slot_id_fkey'),
    sa.ForeignKeyConstraint(['user_id'], ['users.user_id'], name='doctor_bookings_user_id_fkey'),
    sa.PrimaryKeyConstraint('appointment_id', name='doctor_bookings_pkey')
    )
    op.create_table('doctor_slots',
    sa.Column('doctor_slot_id', sa.INTEGER(), autoincrement=True, nullable=False),
    sa.Column('doctor_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.Column('slot_id', sa.INTEGER(), autoincrement=False, nullable=False),
    sa.ForeignKeyConstraint(['doctor_id'], ['doctors.doctor_id'], name='doctor_slots_doctor_id_fkey', ondelete='CASCADE'),
    sa.ForeignKeyConstraint(['slot_id'], ['slots.slot_id'], name='doctor_slots_slot_id_fkey', ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('doctor_slot_id', name='doctor_slots_pkey')
    )
    op.create_foreign_key('bookings_doctor_slot_id_fkey', 'bookings', 'doctor_slots', ['doctor_slot_id'], ['doctor_slot_id'], ondelete='CASCADE')
    op.create_foreign_key('bookings_diagnostic_slot_id_fkey', 'bookings', 'diagnostic_slots', ['diagnostic_slot_id'], ['diagnostic_slot_id'], ondelete='CASCADE')
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/2651025c84a1_removed_slots_completely.py
Latand/MedSyncWebApp
"""change slots and add working hours

Revision ID: 3111b176081c
Revises: f8318189ccce
Create Date: 2023-09-30 14:57:37.482302

"""
import random
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql

# revision identifiers, used by Alembic.
revision: str = "3111b176081c"
down_revision: Union[str, None] = "f8318189ccce"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
working_hours_variants = [
    {"start_time": 8, "end_time": 17},
    {"start_time": 9, "end_time": 18},
    {"start_time": 10, "end_time": 19},
    {"start_time": 11, "end_time": 20},
]


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "workinghours",
        sa.Column("working_hour_id", sa.Integer(), nullable=False),
        sa.Column("location_id", sa.Integer(), nullable=False),
        sa.Column("start_time", sa.Integer(), nullable=False),
        sa.Column("end_time", sa.Integer(), nullable=False),
        sa.Column("weekday_index", sa.Integer(), nullable=False),
        sa.Column(
            "created_at",
            postgresql.TIMESTAMP(),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.ForeignKeyConstraint(
            ["location_id"],
            ["locations.location_id"],
        ),
        sa.PrimaryKeyConstraint("working_hour_id"),
    )
    op.drop_column("diagnostic_slots", "is_booked")
    op.drop_column("doctor_slots", "is_booked")
    op.drop_column("doctors", "working_time")
    op.add_column("slots", sa.Column("date", sa.Date(), nullable=False))
    op.add_column(
        "slots",
        sa.Column(
            "is_booked", sa.Boolean(), server_default=sa.text("false"), nullable=False
        ),
    )
    op.drop_column("slots", "end_time")

    # Get the list of existing location IDs
    conn = op.get_bind()
    result = conn.scalars(sa.text("SELECT location_id FROM locations;"))
    # Insert working hours for each location
    for location_id in result:
        for weekday_index in range(6):  # 0=Monday, 1=Tuesday, ..., 6=Sunday, Sunday is not working day
            # Randomly select working hours variant
            working_hours = random.choice(working_hours_variants)

            conn.execute(
                sa.text(
                    """
                    INSERT INTO workinghours (location_id, start_time, end_time, weekday_index)
                    VALUES (:location_id, :start_time, :end_time, :weekday_index);
                """
                ),
                parameters=dict(
                    location_id=location_id,
                    start_time=working_hours["start_time"],
                    end_time=working_hours["end_time"],
                    weekday_index=weekday_index,
                ),
            )

    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column(
        "slots",
        sa.Column(
            "end_time", postgresql.TIMESTAMP(), autoincrement=False, nullable=False
        ),
    )
    op.drop_column("slots", "is_booked")
    op.drop_column("slots", "date")
    op.add_column(
        "doctors",
        sa.Column("working_time", sa.TEXT(), autoincrement=False, nullable=True),
    )
    op.add_column(
        "doctor_slots",
        sa.Column(
            "is_booked",
            sa.BOOLEAN(),
            server_default=sa.text("false"),
            autoincrement=False,
            nullable=False,
        ),
    )
    op.add_column(
        "diagnostic_slots",
        sa.Column(
            "is_booked",
            sa.BOOLEAN(),
            server_default=sa.text("false"),
            autoincrement=False,
            nullable=False,
        ),
    )
    op.drop_table("workinghours")
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/3111b176081c_change_slots_and_add_working_hours.py
Latand/MedSyncWebApp
"""edit bookings

Revision ID: 54b5265d2e43
Revises: 0b7dc85ebb76
Create Date: 2023-10-01 19:16:39.146289

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

# revision identifiers, used by Alembic.
revision: str = '54b5265d2e43'
down_revision: Union[str, None] = '0b7dc85ebb76'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('bookings', 'user_id',
               existing_type=sa.BIGINT(),
               nullable=True)
    op.alter_column('bookings', 'booking_time',
               existing_type=postgresql.TIMESTAMP(),
               type_=sa.TIMESTAMP(timezone=True),
               existing_nullable=False)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column('bookings', 'booking_time',
               existing_type=sa.TIMESTAMP(timezone=True),
               type_=postgresql.TIMESTAMP(),
               existing_nullable=False)
    conn = op.get_bind()
    conn.execute(sa.text("TRUNCATE bookings RESTART IDENTITY CASCADE"))
    op.alter_column('bookings', 'user_id',
               existing_type=sa.BIGINT(),
               nullable=False)
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/54b5265d2e43_edit_bookings.py
Latand/MedSyncWebApp
"""seed doctor ratings

Revision ID: 566ce5e78aa3
Revises: 2651025c84a1
Create Date: 2023-10-01 08:36:35.653763

"""
import random
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision: str = "566ce5e78aa3"
down_revision: Union[str, None] = "2651025c84a1"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    conn = op.get_bind()

    doctor_ids = conn.scalars(
        sa.text("SELECT doctor_id FROM doctors"),
    )
    bad_review = lambda: random.randint(0, 50) < 2

    for doctor_id in doctor_ids:
        conn.execute(
            sa.text(
                "INSERT INTO doctor_ratings (doctor_id, rating) VALUES (:doctor_id, :rating)"
            ),
            parameters=[
                dict(
                    doctor_id=doctor_id,
                    rating=random.randint(4, 5) if not bad_review() else random.randint(1, 3),
                )
                for _ in range(random.randint(50, 250))
            ],
        )

    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    conn = op.get_bind()
    conn.execute(sa.text("TRUNCATE TABLE doctor_ratings"))
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/566ce5e78aa3_seed_doctor_ratings.py
Latand/MedSyncWebApp
"""seed data

Revision ID: 5a5dafefde9b
Revises: c3601dc5d69d
Create Date: 2023-09-29 07:06:41.042185

"""
import random
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op
from faker import Faker

# revision identifiers, used by Alembic.
revision: str = "5a5dafefde9b"
down_revision: Union[str, None] = "c3601dc5d69d"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # Get the database connection
    conn = op.get_bind()
    # Initialize Faker
    fake = Faker()

    # Seed Locations
    location_ids = []
    for _ in range(10):
        result = conn.execute(
            sa.text(
                "INSERT INTO locations (name, address) VALUES (:name, :address) RETURNING location_id;",
            ),
            parameters=dict(
                name=fake.company(),
                address=fake.address(),
            ),
        )
        location_id = result.scalar()
        location_ids.append(location_id)

    # Seed Diagnostics
    diagnostic_ids = []
    for _ in range(5):
        result = conn.execute(
            sa.text(
                "INSERT INTO diagnostics (type_name, description, price) VALUES (:type_name, :description, :price) RETURNING diagnostic_id;",
            ),
            parameters=dict(
                type_name=fake.bs(),
                description=fake.catch_phrase(),
                price=float("{:.2f}".format(random.uniform(50.0, 500.0))),
            ),
        )
        diagnostic_id = result.scalar()
        diagnostic_ids.append(diagnostic_id)

    # Seed Diagnostic_Locations
    for diag_id in diagnostic_ids:
        for loc_id in random.sample(location_ids, 3):
            conn.execute(
                sa.text(
                "INSERT INTO diagnostic_locations (diagnostic_id, location_id) VALUES (:diagnostic_id, :location_id);",
                ),
                parameters=dict(
                diagnostic_id=diag_id,
                location_id=loc_id,
                )
            )

    # Seed Doctors
    for loc_id in location_ids:
        conn.execute(
            sa.text(
            "INSERT INTO doctors (location_id, full_name, specialty, price, photo_url) VALUES (:location_id, :full_name, :specialty, :price, :photo_url) RETURNING doctor_id;",
            ),
            parameters=dict(
            location_id=loc_id,
            full_name=fake.name(),
            specialty=fake.job(),
            price=float("{:.2f}".format(random.uniform(50.0, 200.0))),
            photo_url=fake.image_url(),
            )
        )



def downgrade() -> None:
    pass

backend/src/medsyncapp/migrations/versions/5a5dafefde9b_seed_data.py
Latand/MedSyncWebApp
"""move specialties

Revision ID: 7b0bae0ba599
Revises: 5a5dafefde9b
Create Date: 2023-09-29 11:56:20.206721

"""
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op

# revision identifiers, used by Alembic.
revision: str = "7b0bae0ba599"
down_revision: Union[str, None] = "5a5dafefde9b"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    conn = op.get_bind()

    conn.execute(
        sa.text(
            "TRUNCATE TABLE doctors, diagnostic_locations, diagnostics, locations RESTART IDENTITY CASCADE;"
        )
    )
    op.create_table(
        "specialties",
        sa.Column("specialty_id", sa.Integer(), nullable=False),
        sa.Column("specialty_name", sa.String(length=128), nullable=False),
        sa.PrimaryKeyConstraint("specialty_id"),
    )
    op.add_column("doctors", sa.Column("specialty_id", sa.Integer(), nullable=False))
    op.create_foreign_key(
        None, "doctors", "specialties", ["specialty_id"], ["specialty_id"]
    )
    op.drop_column("doctors", "specialty")
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column(
        "doctors",
        sa.Column(
            "specialty", sa.VARCHAR(length=128), autoincrement=False, nullable=False
        ),
    )
    op.drop_constraint(None, "doctors", type_="foreignkey")
    op.drop_column("doctors", "specialty_id")
    op.drop_table("specialties")
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/7b0bae0ba599_move_specialties.py
Latand/MedSyncWebApp
"""add doctor about info

Revision ID: 80330b4aa722
Revises: 0e5cf4b909ff
Create Date: 2023-09-29 18:27:35.417696

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = '80330b4aa722'
down_revision: Union[str, None] = '0e5cf4b909ff'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('doctors', sa.Column('experience', sa.TEXT(), nullable=True))
    op.add_column('doctors', sa.Column('certificates', sa.TEXT(), nullable=True))
    op.add_column('doctors', sa.Column('working_time', sa.TEXT(), nullable=True))
    op.add_column('doctors', sa.Column('services', sa.TEXT(), nullable=True))

    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_column('doctors', 'working_time')
    op.drop_column('doctors', 'certificates')
    op.drop_column('doctors', 'experience')
    op.drop_column('doctors', 'services')
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/80330b4aa722_add_doctor_about_info.py
Latand/MedSyncWebApp
"""add diagnostic results

Revision ID: 88bdd79e5a2f
Revises: 0bfc6022b536
Create Date: 2023-10-03 19:30:33.180279

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

# revision identifiers, used by Alembic.
revision: str = '88bdd79e5a2f'
down_revision: Union[str, None] = '0bfc6022b536'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('diagnostic_results',
    sa.Column('booking_id', sa.Integer(), nullable=True),
    sa.Column('diagnostic_result_id', sa.Integer(), nullable=False),
    sa.Column('file_path', sa.String(length=256), nullable=False),
    sa.Column('file_id', sa.String(length=256), nullable=True),
    sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('now()'), nullable=False),
    sa.ForeignKeyConstraint(['booking_id'], ['bookings.booking_id'], ),
    sa.PrimaryKeyConstraint('diagnostic_result_id')
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table('diagnostic_results')
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/88bdd79e5a2f_add_diagnostic_results.py
Latand/MedSyncWebApp
"""add ondelete to slots

Revision ID: a6cf7fd5d39b
Revises: 3111b176081c
Create Date: 2023-09-30 15:08:04.061093

"""
from typing import Sequence, Union

from alembic import op

# revision identifiers, used by Alembic.
revision: str = "a6cf7fd5d39b"
down_revision: Union[str, None] = "3111b176081c"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_constraint(
        "bookings_diagnostic_slot_id_fkey", "bookings", type_="foreignkey"
    )
    op.drop_constraint("bookings_doctor_slot_id_fkey", "bookings", type_="foreignkey")
    op.create_foreign_key(
        None,
        "bookings",
        "diagnostic_slots",
        ["diagnostic_slot_id"],
        ["diagnostic_slot_id"],
        ondelete="CASCADE",
    )
    op.create_foreign_key(
        None,
        "bookings",
        "doctor_slots",
        ["doctor_slot_id"],
        ["doctor_slot_id"],
        ondelete="CASCADE",
    )
    op.drop_constraint(
        "diagnostic_slots_diagnostic_location_id_fkey",
        "diagnostic_slots",
        type_="foreignkey",
    )
    op.drop_constraint(
        "diagnostic_slots_slot_id_fkey", "diagnostic_slots", type_="foreignkey"
    )
    op.create_foreign_key(
        None, "diagnostic_slots", "slots", ["slot_id"], ["slot_id"], ondelete="CASCADE"
    )
    op.create_foreign_key(
        None,
        "diagnostic_slots",
        "diagnostic_locations",
        ["diagnostic_location_id"],
        ["diagnostic_location_id"],
        ondelete="CASCADE",
    )
    op.drop_constraint("doctor_slots_slot_id_fkey", "doctor_slots", type_="foreignkey")
    op.drop_constraint(
        "doctor_slots_doctor_id_fkey", "doctor_slots", type_="foreignkey"
    )
    op.create_foreign_key(
        None, "doctor_slots", "slots", ["slot_id"], ["slot_id"], ondelete="CASCADE"
    )
    op.create_foreign_key(
        None,
        "doctor_slots",
        "doctors",
        ["doctor_id"],
        ["doctor_id"],
        ondelete="CASCADE",
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_foreign_key(
        "doctor_slots_doctor_id_fkey",
        "doctor_slots",
        "doctors",
        ["doctor_id"],
        ["doctor_id"],
    )
    op.create_foreign_key(
        "doctor_slots_slot_id_fkey", "doctor_slots", "slots", ["slot_id"], ["slot_id"]
    )
    op.drop_constraint(None, "diagnostic_slots", type_="foreignkey")
    op.drop_constraint(None, "diagnostic_slots", type_="foreignkey")
    op.create_foreign_key(
        "diagnostic_slots_slot_id_fkey",
        "diagnostic_slots",
        "slots",
        ["slot_id"],
        ["slot_id"],
    )
    op.create_foreign_key(
        "diagnostic_slots_diagnostic_location_id_fkey",
        "diagnostic_slots",
        "diagnostic_locations",
        ["diagnostic_location_id"],
        ["diagnostic_location_id"],
    )
    op.drop_constraint(None, "bookings", type_="foreignkey")
    op.drop_constraint(None, "bookings", type_="foreignkey")
    op.create_foreign_key(
        "bookings_doctor_slot_id_fkey",
        "bookings",
        "doctor_slots",
        ["doctor_slot_id"],
        ["doctor_slot_id"],
    )
    op.create_foreign_key(
        "bookings_diagnostic_slot_id_fkey",
        "bookings",
        "diagnostic_slots",
        ["diagnostic_slot_id"],
        ["diagnostic_slot_id"],
    )
    op.drop_constraint(None, "doctor_slots", type_="foreignkey")
    op.drop_constraint(None, "doctor_slots", type_="foreignkey")
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/a6cf7fd5d39b_add_ondelete_to_slots.py
Latand/MedSyncWebApp
"""init database

Revision ID: c3601dc5d69d
Revises: 
Create Date: 2023-09-29 07:03:04.201178

"""
from typing import Sequence, Union

import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql

# revision identifiers, used by Alembic.
revision: str = "c3601dc5d69d"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "diagnostics",
        sa.Column("diagnostic_id", sa.Integer(), nullable=False),
        sa.Column("type_name", sa.String(length=128), nullable=False),
        sa.Column("description", sa.String(length=256), nullable=False),
        sa.Column("price", sa.DECIMAL(precision=10, scale=2), nullable=False),
        sa.PrimaryKeyConstraint("diagnostic_id"),
    )
    op.create_table(
        "locations",
        sa.Column("location_id", sa.Integer(), nullable=False),
        sa.Column("name", sa.String(length=128), nullable=False),
        sa.Column("address", sa.String(length=256), nullable=False),
        sa.PrimaryKeyConstraint("location_id"),
    )
    op.create_table(
        "users",
        sa.Column("user_id", sa.BIGINT(), autoincrement=False, nullable=False),
        sa.Column("username", sa.String(length=128), nullable=True),
        sa.Column("full_name", sa.String(length=128), nullable=False),
        sa.Column(
            "created_at",
            postgresql.TIMESTAMP(),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("user_id"),
    )
    op.create_table(
        "diagnostic_locations",
        sa.Column("diagnostic_location_id", sa.Integer(), nullable=False),
        sa.Column("diagnostic_id", sa.Integer(), nullable=False),
        sa.Column("location_id", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(
            ["diagnostic_id"],
            ["diagnostics.diagnostic_id"],
        ),
        sa.ForeignKeyConstraint(
            ["location_id"],
            ["locations.location_id"],
        ),
        sa.PrimaryKeyConstraint("diagnostic_location_id"),
    )
    op.create_table(
        "doctors",
        sa.Column("doctor_id", sa.Integer(), nullable=False),
        sa.Column("location_id", sa.Integer(), nullable=False),
        sa.Column("full_name", sa.String(length=128), nullable=False),
        sa.Column("specialty", sa.String(length=128), nullable=False),
        sa.Column("price", sa.DECIMAL(precision=10, scale=2), nullable=False),
        sa.Column("photo_url", sa.String(length=256), nullable=False),
        sa.ForeignKeyConstraint(
            ["location_id"],
            ["locations.location_id"],
        ),
        sa.PrimaryKeyConstraint("doctor_id"),
    )
    op.create_table(
        "slots",
        sa.Column("slot_id", sa.Integer(), nullable=False),
        sa.Column("location_id", sa.Integer(), nullable=False),
        sa.Column("start_time", sa.TIMESTAMP(), nullable=False),
        sa.Column("end_time", sa.TIMESTAMP(), nullable=False),
        sa.Column(
            "created_at",
            postgresql.TIMESTAMP(),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.ForeignKeyConstraint(
            ["location_id"],
            ["locations.location_id"],
        ),
        sa.PrimaryKeyConstraint("slot_id"),
    )
    op.create_table(
        "diagnostic_slots",
        sa.Column("diagnostic_slot_id", sa.Integer(), nullable=False),
        sa.Column("diagnostic_location_id", sa.Integer(), nullable=False),
        sa.Column("slot_id", sa.Integer(), nullable=False),
        sa.Column(
            "is_booked", sa.Boolean(), server_default=sa.text("false"), nullable=False
        ),
        sa.ForeignKeyConstraint(
            ["diagnostic_location_id"],
            ["diagnostic_locations.diagnostic_location_id"],
        ),
        sa.ForeignKeyConstraint(
            ["slot_id"],
            ["slots.slot_id"],
        ),
        sa.PrimaryKeyConstraint("diagnostic_slot_id"),
    )
    op.create_table(
        "doctor_bookings",
        sa.Column("appointment_id", sa.Integer(), nullable=False),
        sa.Column("user_id", sa.BIGINT(), nullable=False),
        sa.Column("doctor_id", sa.Integer(), nullable=False),
        sa.Column("slot_id", sa.Integer(), nullable=False),
        sa.Column(
            "status",
            sa.Enum("Booked", "Cancelled", name="booking_status"),
            nullable=False,
        ),
        sa.Column(
            "created_at",
            postgresql.TIMESTAMP(),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.ForeignKeyConstraint(
            ["doctor_id"],
            ["doctors.doctor_id"],
        ),
        sa.ForeignKeyConstraint(
            ["slot_id"],
            ["slots.slot_id"],
        ),
        sa.ForeignKeyConstraint(
            ["user_id"],
            ["users.user_id"],
        ),
        sa.PrimaryKeyConstraint("appointment_id"),
    )
    op.create_table(
        "doctor_ratings",
        sa.Column("rating_id", sa.Integer(), nullable=False),
        sa.Column("doctor_id", sa.Integer(), nullable=False),
        sa.Column("rating", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(
            ["doctor_id"],
            ["doctors.doctor_id"],
        ),
        sa.PrimaryKeyConstraint("rating_id"),
    )
    op.create_table(
        "doctor_slots",
        sa.Column("doctor_slot_id", sa.Integer(), nullable=False),
        sa.Column("doctor_id", sa.Integer(), nullable=False),
        sa.Column("slot_id", sa.Integer(), nullable=False),
        sa.Column(
            "is_booked", sa.Boolean(), server_default=sa.text("false"), nullable=False
        ),
        sa.ForeignKeyConstraint(
            ["doctor_id"],
            ["doctors.doctor_id"],
        ),
        sa.ForeignKeyConstraint(
            ["slot_id"],
            ["slots.slot_id"],
        ),
        sa.PrimaryKeyConstraint("doctor_slot_id"),
    )
    op.create_table(
        "bookings",
        sa.Column("booking_id", sa.Integer(), nullable=False),
        sa.Column("user_full_name", sa.String(length=256), nullable=False),
        sa.Column("user_email", sa.String(length=256), nullable=False),
        sa.Column("user_phone_number", sa.String(length=16), nullable=False),
        sa.Column("doctor_slot_id", sa.Integer(), nullable=True),
        sa.Column("diagnostic_slot_id", sa.Integer(), nullable=True),
        sa.Column(
            "created_at",
            postgresql.TIMESTAMP(),
            server_default=sa.text("now()"),
            nullable=False,
        ),
        sa.ForeignKeyConstraint(
            ["diagnostic_slot_id"],
            ["diagnostic_slots.diagnostic_slot_id"],
        ),
        sa.ForeignKeyConstraint(
            ["doctor_slot_id"],
            ["doctor_slots.doctor_slot_id"],
        ),
        sa.PrimaryKeyConstraint("booking_id"),
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    # delete enum:
    op.execute("DROP TYPE booking_status")
    op.drop_table("bookings")
    op.drop_table("doctor_slots")
    op.drop_table("doctor_ratings")
    op.drop_table("doctor_bookings")
    op.drop_table("diagnostic_slots")
    op.drop_table("slots")
    op.drop_table("doctors")
    op.drop_table("diagnostic_locations")
    op.drop_table("users")
    op.drop_table("locations")
    op.drop_table("diagnostics")
    # ### end Alembic commands ###

backend/src/medsyncapp/migrations/versions/c3601dc5d69d_init_database.py
Latand/MedSyncWebApp
"""seed doctor about data

Revision ID: f8318189ccce
Revises: 80330b4aa722
Create Date: 2023-09-29 19:01:16.356469

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision: str = 'f8318189ccce'
down_revision: Union[str, None] = '80330b4aa722'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

doctor_about_data = [
    {
        "doctor_id": 1,
        "experience": "With a robust experience in diagnosing and treating a variety of neurological conditions such as epilepsy, Alzheimer's disease, and migraine, I ensure a thorough examination and tailored treatment plans for every patient.",
        "services": "Neurological consultation\nEpilepsy management\nAlzheimer's treatment\nMigraine treatment.",
        "certificates": "Board Certified in Neurology, Fellowship in Epilepsy.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    },
    {
        "doctor_id": 2,
        "experience": "My expertise in diagnosing and treating eye conditions has been honed over years of practice. I specialize in cataract surgery, glaucoma management, and refractive surgery to correct vision problems.",
        "services": "Eye examinations\nCataract surgery\nGlaucoma management\nRefractive surgery.",
        "certificates": "Board Certified in Ophthalmology, Fellowship in Cornea and Refractive Surgery.",
        "working_time": "Monday to Friday: 8:00 AM - 4:00 PM."
    },
    {
        "doctor_id": 3,
        "experience": "As a neurologist with extensive experience, I focus on providing personalized care for patients with neurological disorders, ensuring a comprehensive approach towards diagnosis and treatment.",
        "services": "Neurological consultation\nParkinson's disease management\nStroke rehabilitation.",
        "certificates": "Board Certified in Neurology, Certified in Clinical Neurophysiology.",
        "working_time": "Monday to Friday: 10:00 AM - 6:00 PM."
    },
    {
        "doctor_id": 4,
        "experience": "Specializing in blood disorders and cancers, I am dedicated to providing exceptional care and innovative treatments to patients with a variety of hematological conditions.",
        "services": "Hematology consultation\nBlood disorder treatment\nHematologic oncology.",
        "certificates": "Board Certified in Hematology, Fellowship in Hematologic Oncology.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    },
    {
        "doctor_id": 5,
        "experience": "With a profound knowledge in eye care, I specialize in diagnosing and treating various eye conditions ensuring a clear vision and healthy eyes for my patients.",
        "services": "Eye examinations\nDiabetic retinopathy management\nMacular degeneration treatment.",
        "certificates": "Board Certified in Ophthalmology, Certified in Retinal Diseases.",
        "working_time": "Monday to Friday: 8:00 AM - 4:00 PM."
    },
    {
        "doctor_id": 6,
        "experience": "I am dedicated to managing hormonal disorders and ensuring that my patients receive the appropriate care and treatment to maintain a balanced hormonal health.",
        "services": "Endocrinology consultation\nDiabetes management\nThyroid disorder treatment.",
        "certificates": "Board Certified in Endocrinology, Certified in Diabetes Care.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    },
    {
        "doctor_id": 7,
        "experience": "Providing comprehensive neurological care, I specialize in treating a range of neurological disorders ensuring a better quality of life for my patients.",
        "services": "Neurological consultation\nMultiple Sclerosis management\nHeadache treatment.",
        "certificates": "Board Certified in Neurology, Certified in Neuroimmunology.",
        "working_time": "Monday to Friday: 10:00 AM - 6:00 PM."
    },
    {
        "doctor_id": 8,
        "experience": "I am dedicated to women's health, providing gynecological care that covers a wide range of women's health issues from preventive care to treatment of gynecological disorders.",
        "services": "Gynecological consultation\nPrenatal care\nMenopause management.",
        "certificates": "Board Certified in Obstetrics and Gynecology, Fellowship in Minimally Invasive Gynecologic Surgery.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    },
    {
        "doctor_id": 9,
        "experience": "Specializing in endocrinology, I focus on providing individualized care to manage hormonal disorders, ensuring a balanced and healthier life for my patients.",
        "services": "Endocrinology consultation\nDiabetes management\nThyroid disorder treatment.",
        "certificates": "Board Certified in Endocrinology, Certified in Metabolic Bone Disease.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    },
    {
        "doctor_id": 10,
        "experience": "With a focus on women's health, I provide a range of gynecological services ensuring a comfortable and understanding environment for every patient.",
        "services": "Gynecological consultation\nFamily planning\nReproductive endocrinology.",
        "certificates": "Board Certified in Obstetrics and Gynecology, Certified in Reproductive Endocrinology and Infertility.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    },
    {
        "doctor_id": 11,
        "experience": "Specializing in gastroenterology, I provide comprehensive care for digestive and liver diseases, ensuring accurate diagnosis and effective treatment.",
        "services": "Gastroenterology consultation\nColonoscopy\nLiver disease treatment.",
        "certificates": "Board Certified in Gastroenterology, Fellowship in Hepatology.",
        "working_time": "Monday to Friday: 8:00 AM - 4:00 PM."
    },
    {
        "doctor_id": 12,
        "experience": "As an endocrinologist, I am devoted to treating hormonal imbalances and disorders ensuring the well-being and hormonal health of my patients.",
        "services": "Endocrinology consultation\nDiabetes management\nOsteoporosis treatment.",
        "certificates": "Board Certified in Endocrinology, Certified in Bone Densitometry.",
        "working_time": "Monday to Friday: 9:00 AM - 5:00 PM."
    }
]



def upgrade() -> None:

    # ### commands auto generated by Alembic - please adjust! ###
    conn = op.get_bind()
    for doctor_about in doctor_about_data:
        conn.execute(
            sa.text(
                "UPDATE doctors SET experience = :experience, services = :services, certificates = :certificates, working_time = :working_time WHERE doctor_id = :doctor_id"
            ),
            parameters=doctor_about
        )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    # ### end Alembic commands ###
    pass

backend/src/medsyncapp/migrations/versions/f8318189ccce_seed_doctor_about_data.py
Latand/MedSyncWebApp
import asyncio
import logging

import betterlogging as bl
from aiogram import Bot, Dispatcher
from aiogram.fsm.storage.redis import DefaultKeyBuilder, RedisStorage
from aiogram_dialog import setup_dialogs

from medsyncapp.infrastructure.database.setup import create_engine, create_session_pool
from medsyncapp.tgbot.config import load_config
from medsyncapp.tgbot.handlers import routers_list
from medsyncapp.tgbot.handlers.bookings import booking_dialog
from medsyncapp.tgbot.handlers.test_results import test_results_router, test_results_dialog
from medsyncapp.tgbot.middlewares.database import DatabaseMiddleware
from medsyncapp.tgbot.services import broadcaster


async def on_startup(bot: Bot, admin_ids: list[int]):
    await broadcaster.broadcast(bot, admin_ids, "Bot has been started")


def setup_logging():
    """
    Set up logging configuration for the application.

    This method initializes the logging configuration for the application.
    It sets the log level to INFO and configures a basic colorized log for
    output. The log format includes the filename, line number, log level,
    timestamp, logger name, and log message.

    Returns:
        None

    Example usage:
        setup_logging()
    """
    log_level = logging.INFO
    bl.basic_colorized_config(level=log_level)

    logging.basicConfig(
        level=logging.INFO,
        format="%(filename)s:%(lineno)d #%(levelname)-8s [%(asctime)s] - %(name)s - %(message)s",
    )
    logger = logging.getLogger(__name__)
    logger.info("Starting bot")


async def main():
    setup_logging()

    config = load_config(".env")

    bot = Bot(token=config.tg_bot.token, parse_mode="HTML")
    storage = RedisStorage.from_url(
        config.redis.dsn(),
        key_builder=DefaultKeyBuilder(with_bot_id=True, with_destiny=True),
    )
    dp = Dispatcher(storage=storage)
    engine = create_engine(config.db)
    session_pool = create_session_pool(engine)
    dp.update.outer_middleware(DatabaseMiddleware(session_pool))

    dp.include_routers(*routers_list)
    dp.include_routers(booking_dialog, test_results_dialog)
    setup_dialogs(dp)

    await on_startup(bot, config.tg_bot.admin_ids)
    await dp.start_polling(bot, config=config)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except (KeyboardInterrupt, SystemExit):
        logging.error("Bot has been stopped")

backend/src/medsyncapp/tgbot/bot.py
Latand/MedSyncWebApp
from dataclasses import dataclass
from typing import Optional

from environs import Env


@dataclass
class DbConfig:
    """
    Database configuration class.
    This class holds the settings for the database, such as host, password, port, etc.

    Attributes
    ----------
    host : str
        The host where the database server is located.
    password : str
        The password used to authenticate with the database.
    user : str
        The username used to authenticate with the database.
    database : str
        The name of the database.
    port : int
        The port where the database server is listening.
    """

    host: str
    password: str
    user: str
    database: str
    port: int = 5432

    # For SQLAlchemy
    def construct_sqlalchemy_url(self, driver="asyncpg", host=None, port=None) -> str:
        """
        Constructs and returns a SQLAlchemy URL for this database configuration.
        """
        # TODO: If you're using SQLAlchemy, move the import to the top of the file!
        from sqlalchemy.engine.url import URL

        if not host:
            host = self.host
        if not port:
            port = self.port
        uri = URL.create(
            drivername=f"postgresql+{driver}",
            username=self.user,
            password=self.password,
            host=host,
            port=port,
            database=self.database,
        )
        return uri.render_as_string(hide_password=False)

    @staticmethod
    def from_env(env: Env):
        """
        Creates the DbConfig object from environment variables.
        """
        host = env.str("DB_HOST")
        password = env.str("POSTGRES_PASSWORD")
        user = env.str("POSTGRES_USER")
        database = env.str("POSTGRES_DB")
        port = env.int("DB_PORT", 5432)
        return DbConfig(
            host=host, password=password, user=user, database=database, port=port
        )


@dataclass
class TgBot:
    """
    Creates the TgBot object from environment variables.
    """

    token: str
    admin_ids: list[int]
    use_redis: bool
    web_app_domain: str

    @staticmethod
    def from_env(env: Env):
        """
        Creates the TgBot object from environment variables.
        """
        token = env.str("BOT_TOKEN")
        admin_ids = list(map(int, env.list("ADMINS")))
        use_redis = env.bool("USE_REDIS")
        web_app_domain = env.str("FRONTEND_URL")
        return TgBot(
            token=token,
            admin_ids=admin_ids,
            use_redis=use_redis,
            web_app_domain=web_app_domain,
        )


@dataclass
class RedisConfig:
    """
    Redis configuration class.

    Attributes
    ----------
    redis_pass : Optional(str)
        The password used to authenticate with Redis.
    redis_port : Optional(int)
        The port where Redis server is listening.
    redis_host : Optional(str)
        The host where Redis server is located.
    """

    redis_pass: Optional[str]
    redis_port: Optional[int]
    redis_host: Optional[str]

    def dsn(self) -> str:
        """
        Constructs and returns a Redis DSN (Data Source Name) for this database configuration.
        """
        if self.redis_pass:
            return f"redis://:{self.redis_pass}@{self.redis_host}:{self.redis_port}/0"
        else:
            return f"redis://{self.redis_host}:{self.redis_port}/0"

    @staticmethod
    def from_env(env: Env):
        """
        Creates the RedisConfig object from environment variables.
        """
        redis_pass = env.str("REDIS_PASSWORD")
        redis_port = env.int("REDIS_PORT")
        redis_host = env.str("REDIS_HOST")

        return RedisConfig(
            redis_pass=redis_pass, redis_port=redis_port, redis_host=redis_host
        )


@dataclass
class Config:
    """
    The main configuration class that integrates all the other configuration classes.

    This class holds the other configuration classes, providing a centralized point of access for all settings.

    Attributes
    ----------
    tg_bot : TgBot
        Holds the settings related to the Telegram Bot.
    db : Optional[DbConfig]
        Holds the settings specific to the database (default is None).
    redis : Optional[RedisConfig]
        Holds the settings specific to Redis (default is None).
    """

    tg_bot: TgBot
    db: Optional[DbConfig] = None
    redis: Optional[RedisConfig] = None


def load_config(path: str = None) -> Config:
    """
    This function takes an optional file path as input and returns a Config object.
    :param path: The path of env file from where to load the configuration variables.
    It reads environment variables from a .env file if provided, else from the process environment.
    :return: Config object with attributes set as per environment variables.
    """

    # Create an Env object.
    # The Env object will be used to read environment variables.
    env = Env()
    env.read_env(path)

    return Config(
        tg_bot=TgBot.from_env(env),
        db=DbConfig.from_env(env),
        redis=RedisConfig.from_env(env),
    )

backend/src/medsyncapp/tgbot/config.py
Latand/MedSyncWebApp
"""Import all routers and add them to routers_list."""
from .bookings import profile_router
from .start import start_router
from .test_results import test_results_router
from .web_app import web_app_router

routers_list = [
    start_router,
    profile_router,
    test_results_router,
    web_app_router,
]

__all__ = [
    "routers_list",
]

backend/src/medsyncapp/tgbot/handlers/__init__.py
Latand/MedSyncWebApp
import operator
from typing import Any

from aiogram import Router, F, types
from aiogram.fsm.state import StatesGroup, State
from aiogram_dialog import Window, Dialog, DialogManager, StartMode, ShowMode
from aiogram_dialog.widgets.kbd import ScrollingGroup, Select, Cancel, Back
from aiogram_dialog.widgets.text import Format, Const

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.tgbot.handlers.start import start_from_dialog_menu

profile_router = Router()


class MyBookings(StatesGroup):
    show_list = State()
    show_booking = State()


async def get_bookings(dialog_manager: DialogManager, repo: RequestsRepo, **kwargs):
    bookings = await repo.bookings.get_user_bookings(dialog_manager.event.from_user.id)
    return {
        "bookings": [
            (
                ("👨‍⚕️" if booking.Doctor else "🔬")
                + (
                    f"{booking.Booking.booking_time.strftime('%d %B')}"
                    + (
                        f" - {booking.Diagnostic.type_name}"
                        if booking.Diagnostic
                        else f" - {booking.Doctor.full_name}"
                    )
                ),
                booking.Booking.booking_id,
            )
            for booking in bookings
        ]
    }


async def get_booking_info(dialog_manager: DialogManager, repo: RequestsRepo, **kwargs):
    repo: RequestsRepo = dialog_manager.middleware_data.get("repo")
    booking_id = dialog_manager.dialog_data.get("booking_id")
    booking_info = await repo.bookings.get_booking(int(booking_id))

    booking_time = booking_info.Booking.booking_time.strftime("%d %B %Y, %H:%M UTC")

    appointment_type_text = (
        f"👨‍⚕️ Doctor: {booking_info.Doctor.full_name}\n"
        if booking_info.Doctor
        else f"🔬 Diagnostic: {booking_info.Diagnostic.type_name}\n"
    )

    return {
        "text": (
            f"📋 Booking ID: {booking_info.Booking.booking_id}\n"
            f"{appointment_type_text}"
            f"📆 Date & Time: {booking_time}\n\n"
            f"📍 Location: {booking_info.Location.name}: {booking_info.Location.address}\n\n"
            f"Thank you for choosing our service! If you have any questions or need to reschedule, feel free to reach out. 📞"
        )
    }


async def show_booking(
    callback_query: types.CallbackQuery,
    widget: Any,
    dialog_manager: DialogManager,
    booking_id: str,
):
    dialog_manager.dialog_data.update(booking_id=booking_id)
    await dialog_manager.switch_to(MyBookings.show_booking)


booking_dialog = Dialog(
    Window(
        Const("Here is your booking list. Select one to see details"),
        ScrollingGroup(
            Select(
                Format("{item[0]}"),
                id="s_bookings",
                item_id_getter=operator.itemgetter(1),
                items="bookings",
                on_click=show_booking,
            ),
            id="scroll_bookings",
            width=1,
            height=10,
            hide_on_single_page=True,
        ),
        Cancel(Const("Exit"), on_click=start_from_dialog_menu),
        getter=get_bookings,
        state=MyBookings.show_list,
    ),
    Window(
        Const("Here is your booking details\n\n"),
        Format("{text}"),
        Back(Const("Back")),
        getter=get_booking_info,
        state=MyBookings.show_booking,
    ),
)


@profile_router.callback_query(F.data == "my_bookings")
async def my_bookings(
    callback_query: types.CallbackQuery, dialog_manager: DialogManager
):
    await callback_query.answer()
    await dialog_manager.start(MyBookings.show_list, mode=StartMode.RESET_STACK, show_mode=ShowMode.SEND)

backend/src/medsyncapp/tgbot/handlers/bookings.py
Latand/MedSyncWebApp
from typing import Any

from aiogram import types, Router
from aiogram.filters import CommandStart
from aiogram_dialog import DialogManager

from medsyncapp.tgbot.config import Config
from medsyncapp.tgbot.keyboards.inline import main_menu

start_router = Router()


@start_router.message(CommandStart())
async def send_webapp(message: types.Message, config: Config):
    await message.answer(
        "Welcome to MedSync App!\n\n"
        "You can use our Web App to book an appointment with a doctor, or get tested in one of our clinics.",
        reply_markup=main_menu(domain=config.tg_bot.web_app_domain),
    )


async def start_from_dialog_menu(
    callback_query:types.CallbackQuery, widget: Any, dialog_manager: DialogManager
):
    config: Config = dialog_manager.middleware_data.get("config")
    await callback_query.message.answer(
        "Welcome to MedSync App!\n\n"
        "You can use our Web App to book an appointment with a doctor, or get tested in one of our clinics.",
        reply_markup=main_menu(domain=config.tg_bot.web_app_domain),
    )

backend/src/medsyncapp/tgbot/handlers/start.py
Latand/MedSyncWebApp
import logging
import operator
import os
from typing import Any

from aiogram import Router, F, types
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import FSInputFile
from aiogram_dialog import Window, Dialog, DialogManager, StartMode, ShowMode
from aiogram_dialog.widgets.kbd import ScrollingGroup, Select, Cancel
from aiogram_dialog.widgets.text import Format, Const

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.tgbot.handlers.start import start_from_dialog_menu

test_results_router = Router()

PUBLIC_DIR_PATH = "/src/public"

class MyResult(StatesGroup):
    show_list = State()


async def get_results(dialog_manager: DialogManager, repo: RequestsRepo, **kwargs):
    results = await repo.results.get_results(dialog_manager.event.from_user.id)
    return {
        "results": [
            (
                f"📋 {result.Diagnostic.type_name}: {result.Booking.booking_time.strftime('%d %B %Y')}",
                result.DiagnosticResult.diagnostic_result_id,
            )
            for result in results
        ]
    }


async def show_result(
    callback_query: types.CallbackQuery,
    widget: Any,
    dialog_manager: DialogManager,
    result_id: str,
):
    repo: RequestsRepo = dialog_manager.middleware_data.get("repo")
    result = await repo.results.get_result(int(result_id))
    caption = f"📋 {result.Diagnostic.type_name}: {result.Booking.booking_time.strftime('%d %B %Y')}"
    if result.DiagnosticResult.file_id:
        file = result.DiagnosticResult.file_id
    else:
        full_path = os.path.join(PUBLIC_DIR_PATH, result.DiagnosticResult.file_path)
        logging.info(f"full_path: {full_path}")
        file = FSInputFile(full_path)
    msg = await callback_query.message.answer_document(file, caption=caption)

    if not result.DiagnosticResult.file_id:
        await repo.results.save_file_id(int(result_id), msg.document.file_id)


test_results_dialog = Dialog(
    Window(
        Const("Here is your results list. Select one to see the file"),
        ScrollingGroup(
            Select(
                Format("{item[0]}"),
                id="s_result",
                item_id_getter=operator.itemgetter(1),
                items="results",
                on_click=show_result,
            ),
            id="scroll_results",
            width=1,
            height=10,
            hide_on_single_page=True,
        ),
        Cancel(Const("Exit"), on_click=start_from_dialog_menu),
        getter=get_results,
        state=MyResult.show_list,
    ),
)


@test_results_router.callback_query(F.data == "my_results")
async def my_bookings(
    callback_query: types.CallbackQuery, dialog_manager: DialogManager
):
    await callback_query.answer()
    await dialog_manager.start(
        MyResult.show_list, mode=StartMode.RESET_STACK, show_mode=ShowMode.SEND
    )

backend/src/medsyncapp/tgbot/handlers/test_results.py
Latand/MedSyncWebApp
import json

from aiogram import Router, F, types
from aiogram.utils.markdown import hcode, hbold

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo

web_app_router = Router()


@web_app_router.message(F.web_app_data)
async def bot_echo(message: types.Message, repo: RequestsRepo):
    data = json.loads(message.web_app_data.data)

    if data.get("action") == "booking_confirmed":
        booking_id = data.get("booking_id")
        booking_info = await repo.bookings.get_booking(int(booking_id))

        booking_time = booking_info.Booking.booking_time.strftime("%d %B %Y, %H:%M UTC")

        appointment_type_text =f"👨‍⚕️ Doctor: {hbold(booking_info.Doctor.full_name)}\n" if booking_info.Doctor else f"🔬 Diagnostic: {hbold(booking_info.Diagnostic.type_name)}\n"

        await message.answer(
            text=f"🎉 Congratulations! Your booking is confirmed. 🎉\n\n"
                 f"📋 Booking ID: {hcode(booking_id)}\n"
                    f"{appointment_type_text}"
                 f"📆 Date & Time: {hbold(booking_time)}\n\n"
                 f"📍 Location: {hbold(booking_info.Location.name)}: {hbold(booking_info.Location.address)}\n\n"
                 f"Thank you for choosing our service! If you have any questions or need to reschedule, feel free to reach out. 📞"
        )

backend/src/medsyncapp/tgbot/handlers/web_app.py
Latand/MedSyncWebApp
from aiogram.types import WebAppInfo
from aiogram.utils.keyboard import InlineKeyboardBuilder


def main_menu(domain: str):
    kb = InlineKeyboardBuilder()
    kb.button(text="Main Page", web_app=WebAppInfo(url=domain))
    kb.button(
        text="📅 Book an appointment",
        web_app=WebAppInfo(url=f"{domain}/see_a_doctor"),
    )
    kb.button(
        text="📝 Get tested",
        web_app=WebAppInfo(url=f"{domain}/get_tested"),
    )
    kb.button(text="📋 My bookings", callback_data="my_bookings")
    kb.button(text="📋 My Results", callback_data="my_results")
    kb.adjust(1, 2, 2)
    return kb.as_markup()



backend/src/medsyncapp/tgbot/keyboards/inline.py
Latand/MedSyncWebApp
from typing import Callable, Dict, Any, Awaitable

from aiogram import BaseMiddleware
from aiogram.types import Message

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo


class DatabaseMiddleware(BaseMiddleware):
    def __init__(self, session_pool) -> None:
        self.session_pool = session_pool

    async def __call__(
        self,
        handler: Callable[[Message, Dict[str, Any]], Awaitable[Any]],
        event: Message,
        data: Dict[str, Any],
    ) -> Any:
        async with self.session_pool() as session:
            repo = RequestsRepo(session)
            event_from_user = data.get("event_from_user")
            if not event_from_user:
                return await handler(event, data)

            user = await repo.users.get_or_create_user(
                event_from_user.id,
                event_from_user.full_name,
                event_from_user.username
            )

            data["session"] = session
            data["repo"] = repo
            data["user"] = user

            result = await handler(event, data)
        return result
backend/src/medsyncapp/tgbot/middlewares/database.py
Latand/MedSyncWebApp
import asyncio
import logging
from typing import Union

from aiogram import Bot
from aiogram import exceptions
from aiogram.types import InlineKeyboardMarkup


async def send_message(
    bot: Bot,
    user_id: Union[int, str],
    text: str,
    disable_notification: bool = False,
    reply_markup: InlineKeyboardMarkup = None,
) -> bool:
    """
    Safe messages sender

    :param bot: Bot instance.
    :param user_id: user id. If str - must contain only digits.
    :param text: text of the message.
    :param disable_notification: disable notification or not.
    :param reply_markup: reply markup.
    :return: success.
    """
    try:
        await bot.send_message(
            user_id,
            text,
            disable_notification=disable_notification,
            reply_markup=reply_markup,
        )
    except exceptions.TelegramBadRequest as e:
        logging.error("Telegram server says - Bad Request: chat not found")
    except exceptions.TelegramForbiddenError:
        logging.error(f"Target [ID:{user_id}]: got TelegramForbiddenError")
    except exceptions.TelegramRetryAfter as e:
        logging.error(
            f"Target [ID:{user_id}]: Flood limit is exceeded. Sleep {e.retry_after} seconds."
        )
        await asyncio.sleep(e.retry_after)
        return await send_message(
            bot, user_id, text, disable_notification, reply_markup
        )  # Recursive call
    except exceptions.TelegramAPIError:
        logging.exception(f"Target [ID:{user_id}]: failed")
    else:
        logging.info(f"Target [ID:{user_id}]: success")
        return True
    return False


async def broadcast(
    bot: Bot,
    users: list[Union[str, int]],
    text: str,
    disable_notification: bool = False,
    reply_markup: InlineKeyboardMarkup = None,
) -> int:
    """
    Simple broadcaster.
    :param bot: Bot instance.
    :param users: List of users.
    :param text: Text of the message.
    :param disable_notification: Disable notification or not.
    :param reply_markup: Reply markup.
    :return: Count of messages.
    """
    count = 0
    try:
        for user_id in users:
            if await send_message(
                bot, user_id, text, disable_notification, reply_markup
            ):
                count += 1
            await asyncio.sleep(
                0.05
            )  # 20 messages per second (Limit: 30 messages per second)
    finally:
        logging.info(f"{count} messages successful sent.")

    return count

backend/src/medsyncapp/tgbot/services/broadcaster.py
Latand/MedSyncWebApp
import logging

from fastapi import FastAPI, APIRouter
from starlette.middleware.cors import CORSMiddleware

from medsyncapp.webhook import routers

app = FastAPI()
prefix_router = APIRouter(prefix="/api")

log_level = logging.INFO
log = logging.getLogger(__name__)


app.add_middleware(
    CORSMiddleware,
    # allow_origins=["https://medsync.botfather.dev"],
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

logging.getLogger(__name__).setLevel(logging.INFO)
logging.basicConfig(
    level=logging.INFO,
    format="%(filename)s:%(lineno)d #%(levelname)-8s [%(asctime)s] - %(name)s - %(message)s",
)


for router in [
    routers.doctors.doctor_router,
    routers.diagnostics.diagnostics_router,
    routers.slots.slots_router,
    routers.slots.working_hours_router,
    routers.locations.locations_router,
    routers.doctors.specialties_router,
]:
    prefix_router.include_router(router)

app.include_router(prefix_router)

backend/src/medsyncapp/webhook/app.py
Latand/MedSyncWebApp
from . import booking
from . import diagnostics
from . import doctors
from . import locations
from . import slots

backend/src/medsyncapp/webhook/routers/__init__.py
Latand/MedSyncWebApp
import json
import logging
from contextlib import suppress

import sqlalchemy.exc
from aiogram.utils.markdown import hbold, hcode
from fastapi import Depends, HTTPException
from starlette.requests import Request

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.webhook.utils import (
    get_repo,
    validate_telegram_data,
    bot,
    parse_init_data,
)
from .diagnostics import diagnostics_router
from .doctors import doctor_router


async def get_booking_notification_text(repo, booking_id):
    booking_info = await repo.bookings.get_booking(int(booking_id))

    booking_time = booking_info.Booking.booking_time.strftime("%d %B %Y, %H:%M UTC")

    appointment_type_text = (
        f"👨‍⚕️ Doctor: {hbold(booking_info.Doctor.full_name)}\n"
        if booking_info.Doctor
        else f"🔬 Diagnostic: {hbold(booking_info.Diagnostic.type_name)}\n"
    )

    return (
        f"🎉 Congratulations! Your booking is confirmed. 🎉\n\n"
        f"📋 Booking ID: {hcode(booking_id)}\n"
        f"{appointment_type_text}"
        f"📆 Date & Time: {hbold(booking_time)}\n\n"
        f"📍 Location: {hbold(booking_info.Location.name)}: {hbold(booking_info.Location.address)}\n\n"
        f"Thank you for choosing our service! If you have any questions or need to reschedule, feel free to reach out. 📞"
    )


async def book_slot_endpoint(
    payload: dict, item_type: str, repo: RequestsRepo, parsed_data: dict = None
):
    book_repo = repo.doctors if item_type == "doctor" else repo.diagnostics

    user_info = parsed_data.get('user')
    if user_info:
        user_id = json.loads(user_info).get('id')
    else:
        user_id = None

    try:
        booking_id = await book_repo.book_slot(
            payload, user_id=user_id
        )
    except sqlalchemy.exc.IntegrityError as e:
        await repo.session.rollback()
        if "bookings_user_id_fkey" in str(e) and payload.get("user_id"):
            await repo.users.get_or_create_user(
                user_id=payload.get("user_id"),
                full_name=payload.get("user_name", "")
                + " "
                + payload.get("user_surname", ""),
            )
            booking_id = await book_repo.book_slot(payload, user_id=parsed_data.get("user_id"))
        else:
            raise HTTPException(status_code=400, detail="Invalid payload") from e

    return {"status": "success", "booking_id": booking_id}


@doctor_router.post("/book_slot")
async def book_slot(request: Request, repo: RequestsRepo = Depends(get_repo)):
    data = await request.json()

    init_data = data.get("userInitData")
    if init_data and not validate_telegram_data(init_data):
        raise HTTPException(status_code=400, detail="Invalid initData")
    parsed_data = parse_init_data(init_data)
    result = await book_slot_endpoint(data, "doctor", repo, parsed_data)

    if init_data:
        user_info = parsed_data.get("user")
        if user_info:
            user_id = json.loads(user_info).get("id")
            with suppress():
                notification = await get_booking_notification_text(
                    repo, result["booking_id"]
                )
                await bot.send_message(
                    chat_id=user_id,
                    text=notification,
                    parse_mode="HTML",
                )

    return result


@diagnostics_router.post("/book_slot")
async def book_slot(request: Request, repo: RequestsRepo = Depends(get_repo)):
    data = await request.json()

    init_data = data.get("userInitData")
    if init_data and not validate_telegram_data(init_data):
        raise HTTPException(status_code=400, detail="Invalid initData")

    parsed_data = parse_init_data(init_data)
    result = await book_slot_endpoint(data, "diagnostic", repo, parsed_data)
    # This is just for test purposes, we will create a diagnostic result here straightaway
    await repo.results.create_result(
        result["booking_id"], diagnostic_id=data["diagnostic_id"]
    )

    if init_data:
        user_info = parsed_data.get("user")
        if user_info:
            user_id = json.loads(user_info).get("id")
            with suppress():
                notification = await get_booking_notification_text(
                    repo, result["booking_id"]
                )
                await bot.send_message(
                    chat_id=user_id,
                    text=notification,
                    parse_mode="HTML",
                )

    return result

backend/src/medsyncapp/webhook/routers/booking.py
Latand/MedSyncWebApp
from fastapi import Depends, APIRouter, HTTPException

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.webhook.utils import get_repo

diagnostics_router = APIRouter(prefix="/diagnostics")


@diagnostics_router.get("/")
async def get_all_diagnostics(repo: RequestsRepo = Depends(get_repo)):
    diagnostics = await repo.diagnostics.get_all_diagnostic_types()
    return diagnostics


@diagnostics_router.get(
    "/{diagnostic_id}/locations"
)
async def get_locations_by_type(
    diagnostic_id: int, repo: RequestsRepo = Depends(get_repo)
):
    locations = await repo.diagnostics.get_locations_by_type(diagnostic_id)
    if not locations:
        raise HTTPException(status_code=404, detail="Locations not found")
    return locations


backend/src/medsyncapp/webhook/routers/diagnostics.py
Latand/MedSyncWebApp
from fastapi import Depends, APIRouter, HTTPException

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.webhook.utils import get_repo

doctor_router = APIRouter(prefix="/doctors")
specialties_router = APIRouter(prefix="/specialties")


@doctor_router.get("/{doctor_id}")
async def get_doctor(doctor_id: int, repo: RequestsRepo = Depends(get_repo)):
    result = await repo.doctors.get_doctor(doctor_id)

    if result is None:
        raise HTTPException(status_code=404, detail="Doctor not found")

    return result


@doctor_router.get("/")
async def get_all_doctors(repo: RequestsRepo = Depends(get_repo)):
    doctors = await repo.doctors.get_all_doctors()
    return doctors


@specialties_router.get("/")
async def get_specialties(repo: RequestsRepo = Depends(get_repo)):
    specialties = await repo.doctors.get_specialties()
    return specialties

backend/src/medsyncapp/webhook/routers/doctors.py
Latand/MedSyncWebApp
from fastapi import Depends, APIRouter, HTTPException

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.webhook.utils import get_repo

locations_router = APIRouter(prefix="/locations")


@locations_router.get("/{location_id}")
async def get_location(location_id: int, repo: RequestsRepo = Depends(get_repo)):
    location = await repo.locations.get_location(location_id)
    if location is None:
        raise HTTPException(status_code=404, detail="Location not found")
    return location

backend/src/medsyncapp/webhook/routers/locations.py
Latand/MedSyncWebApp
from fastapi import Depends, APIRouter

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.webhook.utils import get_repo

slots_router = APIRouter(prefix="/slots")
working_hours_router = APIRouter(prefix="/working_hours")


@slots_router.get("/doctors/{doctor_id}/{location_id}/{month_number}")
async def get_available_slots(
    doctor_id: int,
    location_id: int,
    month_number: int,
    repo: RequestsRepo = Depends(get_repo),
):
    slots = await repo.doctors.get_booked_slots(doctor_id, location_id, month_number)
    if not slots:
        return []
    return slots


@slots_router.get("/diagnostics/{diagnostic_id}/{location_id}/{month_number}")
async def get_available_slots_diagnostic(
    diagnostic_id: int,
    location_id: int,
    month_number: int,
    repo: RequestsRepo = Depends(get_repo),
):
    slots = await repo.diagnostics.get_booked_slots(
        diagnostic_id, location_id, month_number
    )
    if not slots:
        return []
    return slots


@working_hours_router.get("/{location_id}")
async def get_working_hours(
    location_id: int,
    repo: RequestsRepo = Depends(get_repo),
):
    working_hours = await repo.slots.get_working_hours(location_id)
    if not working_hours:
        return []
    return working_hours

backend/src/medsyncapp/webhook/routers/slots.py
Latand/MedSyncWebApp
import hashlib
import hmac
import logging
import time
from urllib.parse import unquote, parse_qsl

from aiogram import Bot

from medsyncapp.infrastructure.database.repo.requests import RequestsRepo
from medsyncapp.infrastructure.database.setup import create_session_pool, create_engine
from medsyncapp.tgbot.config import load_config, Config

config: Config = load_config()
engine = create_engine(config.db)
session_pool = create_session_pool(engine)
bot = Bot(config.tg_bot.token)


async def get_repo():
    async with session_pool() as session:
        yield RequestsRepo(session)


def parse_init_data(init_data: str = None) -> dict:
    if not init_data:
        return {}

    parsed_data = dict(parse_qsl(init_data))
    return parsed_data


def validate_telegram_data(init_data: str) -> bool:
    parsed_data = parse_init_data(init_data)

    received_hash = parsed_data.pop("hash")
    # Constructing the data-check-string
    fields = sorted(
        [(key, unquote(value)) for key, value in parsed_data.items() if key != "hash"]
    )
    # Constructing the data-check-string using the sorted order
    data_check_string = "\n".join(f"{k}={v}" for k, v in fields)

    print(data_check_string)
    # Computing the secret key
    secret_key = hmac.new(
        b"WebAppData", config.tg_bot.token.encode(), hashlib.sha256
    ).digest()

    # Comparing received hash with computed hash
    computed_hash = hmac.new(
        secret_key, data_check_string.encode(), hashlib.sha256
    ).hexdigest()

    if computed_hash != received_hash:
        return False


    return True

backend/src/medsyncapp/webhook/utils.py
Latand/MedSyncWebApp