Skip to content

API

API

Main.py

create_employee(employee, db=Depends(get_db)) async

Create a new employee.

Parameters:

  • employee (EmployeeCreate): The employee data to create.
  • db (Session, optional): Database session provided by dependency injection.

Returns: - Employee: The newly created employee's details.

Source code in myapp/api/main.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@app.post("/employees/", response_model=Employee)
async def create_employee(employee: EmployeeCreate, db: Session = Depends(get_db)):
    """
    Create a new employee.

    **Parameters:**

    - `employee (EmployeeCreate):` The employee data to create.
    - `db (Session, optional):` Database session provided by dependency injection.

    **Returns:**
        - `Employee:` The newly created employee's details.
    """
    db_employee = EmployeeDB(
        first_name=employee.first_name,
        last_name=employee.last_name,
        email=employee.email,
        salary=employee.salary
    )
    db.add(db_employee)
    db.commit()
    db.refresh(db_employee)
    return Employee(
        employee_id=db_employee.employee_id,
        first_name=db_employee.first_name,
        last_name=db_employee.last_name,
        email=db_employee.email,
        salary=db_employee.salary
    )

delete_employee(employee_id, db=Depends(get_db)) async

Delete an employee by their unique ID.

Parameters:

  • employee_id (int): The unique identifier of the employee to delete.
  • db (Session, optional): Database session provided by dependency injection.

Returns: - dict: A message confirming successful deletion.

Raises:

- `HTTPException:` If the employee is not found, raises a 404 error.
Source code in myapp/api/main.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@app.delete("/employees/{employee_id}")
async def delete_employee(employee_id: int, db: Session = Depends(get_db)):
    """
    Delete an employee by their unique ID.

    **Parameters:**

    - `employee_id (int):` The unique identifier of the employee to delete.
    - `db (Session, optional):` Database session provided by dependency injection.

    **Returns:**
        - `dict:` A message confirming successful deletion.

    **Raises:**

        - `HTTPException:` If the employee is not found, raises a 404 error.
    """
    employee = db.query(EmployeeDB).filter(EmployeeDB.employee_id == employee_id).first()
    if not employee:
        raise HTTPException(status_code=404, detail="Employee not found")
    db.delete(employee)
    db.commit()
    return {"message": "Employee deleted successfully"}

get_employee(employee_id, db=Depends(get_db)) async

Retrieve an employee by their unique ID.

Parameters:

  • employee_id (int): The unique identifier of the employee.
  • db (Session, optional): Database session provided by dependency injection.

Returns: - Employee: The employee's details.

Raises:

- `HTTPException:` If the employee is not found, raises a 404 error.
Source code in myapp/api/main.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@app.get("/employees/{employee_id}", response_model=Employee)
async def get_employee(employee_id: int, db: Session = Depends(get_db)):
    """
    Retrieve an employee by their unique ID.

    **Parameters:**

    - `employee_id (int):` The unique identifier of the employee.
    - `db (Session, optional):` Database session provided by dependency injection.

    **Returns:**
        - `Employee:` The employee's details.

    **Raises:**

        - `HTTPException:` If the employee is not found, raises a 404 error.
    """
    employee = db.query(EmployeeDB).filter(EmployeeDB.employee_id == employee_id).first()
    if employee is None:
        raise HTTPException(status_code=404, detail="Employee not found")
    return employee

update_employee(employee_id, updated_employee, db=Depends(get_db)) async

Update an existing employee's details.

Parameters:

  • employee_id (int): The unique identifier of the employee to update.
  • updated_employee (EmployeeCreate): The new employee data.
  • db (Session, optional): Database session provided by dependency injection.

Returns: - Employee: The updated employee's details.

Raises:

- `HTTPException:` If the employee is not found, raises a 404 error.
Source code in myapp/api/main.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@app.put("/employees/{employee_id}", response_model=Employee)
async def update_employee(employee_id: int, updated_employee: EmployeeCreate, db: Session = Depends(get_db)):
    """
    Update an existing employee's details.

    **Parameters:**

    - `employee_id (int):` The unique identifier of the employee to update.
    - `updated_employee (EmployeeCreate):` The new employee data.
    - `db (Session, optional):` Database session provided by dependency injection.

    **Returns:**
        - `Employee:` The updated employee's details.

    **Raises:**

        - `HTTPException:` If the employee is not found, raises a 404 error.
    """
    employee = db.query(EmployeeDB).filter(EmployeeDB.employee_id == employee_id).first()
    if not employee:
        raise HTTPException(status_code=404, detail="Employee not found")
    for key, value in updated_employee.dict().items():
        setattr(employee, key, value)
    db.commit()
    db.refresh(employee)
    return employee

Models.py

Database Models for the ETL Process.

This module defines the database models using SQLAlchemy for employees, customers, and products.

Modules:

Name Description
- sqlalchemy

For ORM and database schema definition.
- pydantic

For data validation (not used in these models).

CustomerDB

Bases: Base

Represents a Customer in the database.

Attributes:

  • customer_id (int): The unique identifier for the customer (auto-incremented).
  • customer_name (str): The name of the customer.
  • address (str): The address of the customer.
  • city (str): The city where the customer resides.
  • zip_code (str): The customer's postal code.
Source code in myapp/api/Database/models.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
class CustomerDB(Base):
    """
    Represents a Customer in the database.

    **Attributes:**

    - `customer_id (int):` The unique identifier for the customer (auto-incremented).
    - `customer_name (str):` The name of the customer.
    - `address (str):` The address of the customer.
    - `city (str):` The city where the customer resides.
    - `zip_code (str):` The customer's postal code.
    """
    __tablename__ = "customers"
    customer_id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    customer_name = Column(String, index=True)
    address = Column(String)
    city = Column(String)
    zip_code = Column(String)

EmployeeDB

Bases: Base

Represents an Employee in the database.

Attributes:

  • employee_id (int): The unique identifier for the employee (auto-incremented).
  • first_name (str): The first name of the employee.
  • last_name (str): The last name of the employee.
  • email (str): The unique email address of the employee.
  • salary (int): The employee's salary.
Source code in myapp/api/Database/models.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class EmployeeDB(Base):
    """
    Represents an Employee in the database.

    **Attributes:**

    - `employee_id (int):` The unique identifier for the employee (auto-incremented).
    - `first_name (str):` The first name of the employee.
    - `last_name (str):` The last name of the employee.
    - `email (str):` The unique email address of the employee.
    - `salary (int):` The employee's salary.
    """
    __tablename__ = "employees"
    employee_id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    first_name = Column(String, index=True)
    last_name = Column(String, index=True)
    email = Column(String, unique=True, index=True)
    salary = Column(Integer)

ProductDB

Bases: Base

Represents a Product in the database.

Attributes:

  • product_id (int): The unique identifier for the product (auto-incremented).
  • product_name (str): The name of the product.
  • price (float): The price of the product.
  • description (str): A short description of the product.
  • category (str): The category the product belongs to.
Source code in myapp/api/Database/models.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
class ProductDB(Base):
    """
    Represents a Product in the database.

    **Attributes:**

    - `product_id (int):` The unique identifier for the product (auto-incremented).
    - `product_name (str):` The name of the product.
    - `price (float):` The price of the product.
    - `description (str):` A short description of the product.
    - `category (str):` The category the product belongs to.
    """
    __tablename__ = "products"
    product_id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    product_name = Column(String, index=True)
    price = Column(Float)
    description = Column(String)
    category = Column(String)