Skip to content

APP

Streamlit Application for Employee Management.

This module provides a user interface for interacting with a FastAPI backend to perform CRUD operations on employee records, including retrieving, creating, updating, and deleting.

Modules:

Name Description
- streamlit

For creating the web application interface.

- requests

For making HTTP requests to the FastAPI backend.

- os

For accessing environment variables (used for the API URL).

create_employee(first_name, last_name, email, salary)

Create a new employee record.

Parameters:

  • first_name (str): Employee's first name.
  • last_name (str): Employee's last name.
  • email (str): Employee's email address.
  • salary (int): Employee's salary.

Returns: - dict: The created employee's details or an error message if the operation fails.

Source code in myapp/app/app.py
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def create_employee(first_name: str, last_name: str, email: str, salary: int) -> dict:
    """
    Create a new employee record.

    **Parameters:**

    - `first_name (str):` Employee's first name.
    - `last_name (str):` Employee's last name.
    - `email (str):` Employee's email address.
    - `salary (int):` Employee's salary.

    **Returns:**
        - `dict:` The created employee's details or an error message if the operation fails.
    """
    payload = {
        "first_name": first_name,
        "last_name": last_name,
        "email": email,
        "salary": salary
    }
    response = requests.post(f"{api_url}/employees/", json=payload)
    if response.status_code == 200:
        return response.json()
    else:
        st.error("Failed to create employee.")
        return None

delete_employee(employee_id)

Delete an employee record by ID.

Parameters:

  • employee_id (int): The ID of the employee to delete.

Returns: - dict: A success message if the deletion is successful, or an error message if the operation fails.

Source code in myapp/app/app.py
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def delete_employee(employee_id: int) -> dict:
    """
    Delete an employee record by ID.

    **Parameters:**

    - `employee_id (int):` The ID of the employee to delete.

    **Returns:**
        - `dict:` A success message if the deletion is successful, or an error message if the operation fails.
    """
    response = requests.delete(f"{api_url}/employees/{employee_id}")
    if response.status_code == 200:
        return {"message": "Employee deleted successfully."}
    else:
        st.error("Failed to delete employee.")
        return None

get_employee_by_id(employee_id)

Fetch employee details by their ID.

Parameters:

  • employee_id (int): The ID of the employee to retrieve.

Returns: - dict: The employee's details if found, or None with an error message.

Source code in myapp/app/app.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def get_employee_by_id(employee_id: int) -> dict:
    """
    Fetch employee details by their ID.

    **Parameters:**

    - `employee_id (int):` The ID of the employee to retrieve.

    **Returns:**
        - `dict:` The employee's details if found, or `None` with an error message.
    """
    response = requests.get(f"{api_url}/employees/{employee_id}")
    if response.status_code == 200:
        return response.json()
    else:
        st.error("Employee not found.")
        return None

update_salary(employee_id, new_salary)

Update an existing employee's salary by sending the full employee payload.

Parameters:

  • employee_id (int): The ID of the employee whose salary will be updated.
  • new_salary (int): The new salary amount.

Returns: - dict: The updated employee details if successful, or an error message if the operation fails.

Source code in myapp/app/app.py
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def update_salary(employee_id: int, new_salary: int) -> dict:
    """
    Update an existing employee's salary by sending the full employee payload.

    **Parameters:**

    - `employee_id (int):` The ID of the employee whose salary will be updated.
    - `new_salary (int):` The new salary amount.

    **Returns:**
        - `dict:` The updated employee details if successful, or an error message if the operation fails.
    """
    employee = get_employee_by_id(employee_id)
    if not employee:
        st.error("Employee not found, cannot update salary.")
        return None

    payload = {
        "first_name": employee["first_name"],
        "last_name": employee["last_name"],
        "email": employee["email"],
        "salary": new_salary
    }
    response = requests.put(f"{api_url}/employees/{employee_id}", json=payload)

    if response.status_code == 200:
        return response.json()
    else:
        st.error("Failed to update salary.")
        return None