Merge branch 'charles/readme-for-ground-station-cli' of github.com:UWOrbital/OBC-firmware into charles/readme-for-ground-station-cli

Author: c4bae

gs/backend/api/v1/mcc/endpoints/main_commands.py

@@ -1,18 +1,17 @@
-from fastapi import APIRouter, Depends
-from sqlmodel import Session, select
+from fastapi import APIRouter
 
 from gs.backend.api.v1.mcc.models.responses import MainCommandsResponse
-from gs.backend.data.database.engine import get_db_session
-from gs.backend.data.tables.main_tables import MainCommand
+from gs.backend.data.data_wrappers.mcc_wrappers.main_command_wrapper import get_all_main_commands
 
 main_commands_router = APIRouter(tags=["MCC", "Main Commands"])
 
 
 @main_commands_router.get("/")
-async def get_main_commands(db_session: Session = Depends(get_db_session)) -> MainCommandsResponse:
+async def get_main_commands() -> MainCommandsResponse:
     """
-    @brief Gets the main commands that are available for the MCC
+    Gets the main commands that are available for the MCC
+
+    :return: list of all commands
     """
-    main_commands_query = select(MainCommand)
-    items = list(db_session.exec(main_commands_query).all())
+    items = get_all_main_commands()
     return MainCommandsResponse(data=items)

gs/backend/data/data_wrappers/aro_wrapper/aro_request_wrapper.py

@@ -0,0 +1,73 @@
+from datetime import datetime
+from decimal import Decimal
+from uuid import UUID
+
+from sqlmodel import select
+
+from gs.backend.data.database.engine import get_db_session
+from gs.backend.data.enums.aro_requests import ARORequestStatus
+from gs.backend.data.tables.transactional_tables import ARORequest
+
+
+def get_all_requests() -> list[ARORequest]:
+    """
+    Get all the requests from aro
+    """
+    with get_db_session() as session:
+        requests = list(session.exec(select(ARORequest)).all())
+        return requests
+
+
+def add_request(
+    aro_id: UUID,
+    long: Decimal,
+    lat: Decimal,
+    created_on: datetime,
+    request_sent_obc: datetime,
+    taken_date: datetime,
+    transmission: datetime,
+    status: ARORequestStatus,
+) -> ARORequest:
+    """
+    Add a request
+
+    :param long: the longitude represented as a decimal of max 3 decimal places
+    :param lat: the latitude represented as a decimal of max 3 decimal places
+    :param created_on: datetime object representing the date this request was made. defaults to now
+    :param taken_date: datetime object representing the date that this picture was taken on
+    :param taken_date: datetime object representing the date that this picture was trasmitted
+    :param status: the status of the request, can only be from the requets in ARORequestStatus
+    """
+    with get_db_session() as session:
+        request = ARORequest(
+            aro_id=aro_id,
+            latitude=lat,
+            longitude=long,
+            created_on=created_on,
+            request_sent_to_obc_on=request_sent_obc,
+            taken_date=taken_date,
+            transmission=transmission,
+            status=status,
+        )
+
+        session.add(request)
+        session.commit()
+        session.refresh(request)
+        return request
+
+
+def delete_request_by_id(request_id: str) -> list[ARORequest]:
+    """
+    Delete a request based on id
+
+    :param request_id: unique identifier of the request
+    """
+    with get_db_session() as session:
+        request = session.get(ARORequest, request_id)
+        if request:
+            session.delete(request)
+            session.commit()
+        else:
+            raise ValueError("Request not found, ID does not exist")
+
+        return get_all_requests()

gs/backend/data/data_wrappers/aro_wrapper/aro_user_auth_token_wrapper.py

@@ -0,0 +1,54 @@
+from datetime import datetime
+from uuid import UUID
+
+from sqlmodel import select
+
+from gs.backend.data.database.engine import get_db_session
+from gs.backend.data.enums.aro_auth_token import AROAuthToken as AROEnums
+from gs.backend.data.tables.aro_user_tables import AROUserAuthToken
+
+
+def get_all_auth_tokens() -> list[AROUserAuthToken]:
+    """
+    Get all the auth tokens
+    """
+    with get_db_session() as session:
+        auth_tokens = list(session.exec(select(AROUserAuthToken)).all())
+        return auth_tokens
+
+
+def add_auth_token(token: str, user_data_id: UUID, expiry: datetime, auth_type: AROEnums) -> AROUserAuthToken:
+    """
+    Add auth token to the db
+
+    :param token: the auth token we want to add
+    :param user_data_id: UUID which identifies which user this auth token belongs to
+    :param expiry: the date in which this token expires
+    :param auth_type: the type of auth token this is, can only be from AROAuthToken
+    """
+    with get_db_session() as session:
+        auth_token = AROUserAuthToken(token=token, user_data_id=user_data_id, expiry=expiry, auth_type=auth_type)
+
+        session.add(auth_token)
+        session.commit()
+        session.refresh(auth_token)
+        return auth_token
+
+
+def delete_auth_token_by_id(token_id: UUID) -> list[AROUserAuthToken]:
+    """
+    Delete the auth token based on the token id
+
+    :param token_id: the unique identifier for a particular auth token
+    """
+
+    with get_db_session() as session:
+        auth_token = session.get(AROUserAuthToken, token_id)
+
+        if auth_token:
+            session.delete(auth_token)
+            session.commit()
+        else:
+            print("Token does not exist")
+
+        return get_all_auth_tokens()

gs/backend/data/data_wrappers/aro_wrapper/aro_user_data_wrapper.py

@@ -0,0 +1,64 @@
+from uuid import UUID
+
+from sqlmodel import select
+
+from gs.backend.data.database.engine import get_db_session
+from gs.backend.data.tables.aro_user_tables import AROUsers
+
+
+# selects all objects of type AROUser from db and returns them in list
+def get_all_users() -> list[AROUsers]:
+    """
+    Gets all user
+    """
+    with get_db_session() as session:
+        users = list(session.exec(select(AROUsers)).all())
+        return users
+
+
+# adds user to database of type AROUser then fetches the user from database
+# so that the user now has an assigned ID
+def add_user(call_sign: str, email: str, f_name: str, l_name: str, phone_number: str) -> AROUsers:
+    """
+    Add a new user to the AROUser table in database
+
+    :param call_sign: a 6 character string such as ABCDEF
+    :param email: unique email which is bound to the user
+    :param f_name: first name of user
+    :param l_name: last name of user
+    :param phone_numer: phone number of user
+    """
+    with get_db_session() as session:
+        # check if the user already exists with email as it is unique
+        existing_user = session.exec(select(AROUsers).where(AROUsers.email == email)).first()
+
+        if existing_user:
+            raise ValueError("User already exsits based on email")
+
+        user = AROUsers(
+            call_sign=call_sign, email=email, first_name=f_name, last_name=l_name, phone_number=phone_number
+        )
+
+        session.add(user)
+        session.commit()
+        session.refresh(user)
+        return user
+
+
+# deletes the user with given id and returns the remaining users
+def delete_user_by_id(userid: UUID) -> list[AROUsers]:
+    """
+    Use the user.id to delete a user from table
+
+    :param userid: identifier unique to the user
+    """
+    with get_db_session() as session:
+        user = session.get(AROUsers, userid)
+
+        if user:
+            session.delete(user)
+            session.commit()
+        else:
+            raise ValueError("User ID does not exist")
+
+        return get_all_users()

gs/backend/data/data_wrappers/aro_wrapper/aro_user_login_wrapper.py

@@ -0,0 +1,63 @@
+from uuid import UUID
+
+from sqlmodel import select
+
+from gs.backend.data.database.engine import get_db_session
+from gs.backend.data.tables.aro_user_tables import AROUserLogin
+
+
+def get_all_logins() -> list[AROUserLogin]:
+    """
+    Gets all the logins
+    """
+    with get_db_session() as session:
+        user_logins = list(session.exec(select(AROUserLogin)).all())
+        return user_logins
+
+
+def add_login(email: str, pwd: str, hash_algo: str, user_data_id: UUID, email_verification_token: str) -> AROUserLogin:
+    """
+    Add a new user login
+
+    :param email: the email which the user used to sign up
+    :param pwd: the password the user set
+    :hash_algo: the hashing algorithm used to encrypt the password w the salt
+    :user_data_id: the unique identifier which binds this login to the user which created it
+    :email_verification_token: email verification token
+    """
+    with get_db_session() as session:
+        # check if the user exists already
+        existing_login = session.exec(select(AROUserLogin).where(AROUserLogin.email == email)).first()
+
+        if existing_login:
+            raise ValueError("User login already exists based on email")
+
+        user_login = AROUserLogin(
+            email=email,
+            password=pwd,
+            hashing_algorithm_name=hash_algo,
+            user_data_id=user_data_id,
+            email_verification_token=email_verification_token,
+        )
+
+        session.add(user_login)
+        session.commit()
+        session.refresh(user_login)
+        return user_login
+
+
+def delete_login_by_id(loginid: UUID) -> list[AROUserLogin]:
+    """
+    Use the .id to delete a user from table
+
+    :param loginid: unique identifier of the target login
+    """
+    with get_db_session() as session:
+        user_login = session.get(AROUserLogin, loginid)
+        if user_login:
+            session.delete(user_login)
+            session.commit()
+        else:
+            raise ValueError("Login ID does not exist")
+
+        return get_all_logins()

gs/backend/data/data_wrappers/mcc_wrappers/commands_wrapper.py

@@ -0,0 +1,49 @@
+from typing import Any
+from uuid import UUID
+
+from sqlmodel import select
+
+from gs.backend.data.database.engine import get_db_session
+from gs.backend.data.tables.transactional_tables import Commands
+
+
+def get_all_commands() -> list[Commands]:
+    """
+    Get all data wrapper for Commands
+
+    :return: a list of all commands
+    """
+    with get_db_session() as session:
+        commands = list(session.exec(select(Commands)).all())
+        return commands
+
+
+def create_commands(command_data: dict[str, Any]) -> Commands:
+    """
+    Post data wrapper for Commands
+
+    :param command_data: the JSON object of the command to be created
+    :return: the newly created command
+    """
+    with get_db_session() as session:
+        command = Commands(**command_data)
+        session.add(command)
+        session.commit()
+        session.refresh(command)
+        return command
+
+
+def delete_commands_by_id(command_id: UUID) -> Commands:
+    """
+    Delete data wrapper for Commands
+
+    :param command_id: UUID of command to be deleted
+    :return: the deleted command
+    """
+    with get_db_session() as session:
+        command = session.get(Commands, command_id)
+        if not command:
+            raise ValueError("Command not found.")
+        session.delete(command)
+        session.commit()
+        return command

gs/backend/data/data_wrappers/mcc_wrappers/comms_session_wrapper.py

@@ -0,0 +1,49 @@
+from typing import Any
+from uuid import UUID
+
+from sqlmodel import select
+
+from gs.backend.data.database.engine import get_db_session
+from gs.backend.data.tables.transactional_tables import CommsSession
+
+
+def get_all_comms_sessions() -> list[CommsSession]:
+    """
+    Get all data wrapper for CommsSession
+
+    :return: a list of all sessions
+    """
+    with get_db_session() as session:
+        sessions = list(session.exec(select(CommsSession)).all())
+        return sessions
+
+
+def create_comms_session(session_data: dict[str, Any]) -> CommsSession:
+    """
+    Post data wrapper for CommsSession
+
+    :param session_data: the JSON object of the comms_session to be created
+    :return: the newly created comms_session
+    """
+    with get_db_session() as session:
+        comms_session = CommsSession(**session_data)
+        session.add(comms_session)
+        session.commit()
+        session.refresh(comms_session)
+        return comms_session
+
+
+def delete_telemetry_by_id(session_id: UUID) -> CommsSession:
+    """
+    Delete data wrapper for CommsSession
+
+    :param session_id: UUID of session to be deleted
+    :return: the deleted session
+    """
+    with get_db_session() as session:
+        comms_session = session.get(CommsSession, session_id)
+        if not comms_session:
+            raise ValueError("Comms session not found.")
+        session.delete(comms_session)
+        session.commit()
+        return comms_session