Base

Bases: DeclarativeBase

Base class for SQLAlchemy ORM models.

Source code in api/database.py 
34
35
36
class Base(DeclarativeBase):
    """Base class for SQLAlchemy ORM models."""
    pass

DBpredictions

Bases: Base

ORM model for the 'predictions' table.

Attributes:
  • prediction_id (str) –

    Primary key, unique identifier for the prediction.

  • timestamp (str) –

    Timestamp of the prediction.

  • produit_recu (int) –

    Quantity of the received product.

  • temps_livraison (int) –

    Delivery time.

  • prediction (int) –

    The prediction value.

  • model (str) –

    The model used for the prediction.
Source code in api/database.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class DBpredictions(Base):
    """
    ORM model for the 'predictions' table.

    Attributes:
        prediction_id (str): Primary key, unique identifier for the prediction.
        timestamp (str): Timestamp of the prediction.
        produit_recu (int): Quantity of the received product.
        temps_livraison (int): Delivery time.
        prediction (int): The prediction value.
        model (str): The model used for the prediction.
    """
    __tablename__ = "predictions"

    prediction_id: Mapped[str] = mapped_column(primary_key=True, index=True)
    timestamp: Mapped[str]
    produit_recu: Mapped[int]
    temps_livraison: Mapped[int]
    prediction: Mapped[int]
    model: Mapped[str]

connect_to_postgres()

Connect to a PostgreSQL database using environment variables.

Returns:
  • engine –

    SQLAlchemy engine instance for the PostgreSQL connection.
Source code in api/database.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def connect_to_postgres(): 
    """
    Connect to a PostgreSQL database using environment variables.

    Returns:
        engine: SQLAlchemy engine instance for the PostgreSQL connection.
    """
    load_dotenv()
    # Define your PostgreSQL connection parameters
    hostname = os.environ.get("SERVER")
    database = os.environ.get("DATABASE")
    username = os.environ.get("POSTGRES_USER")
    password = os.environ.get("PASSWORD")

    # Create a connection to the PostgreSQL database
    connection_string = f"postgresql://{username}:{password}@{hostname}/{database}"
    print(connection_string)
    engine = create_engine(connection_string)
    print(engine)
    return engine

create_db_prediction(prediction, session)

Create a new prediction record in the database.

Parameters:
  • prediction (dict) –

    A dictionary containing prediction details.

  • session (Session) –

    SQLAlchemy session object.
Returns:
  • DBpredictions( DBpredictions ) –

    The newly created DBpredictions object.
Source code in api/database.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def create_db_prediction(prediction: dict, session: Session) -> DBpredictions:
    """
    Create a new prediction record in the database.

    Args:
        prediction (dict): A dictionary containing prediction details.
        session (Session): SQLAlchemy session object.

    Returns:
        DBpredictions: The newly created DBpredictions object.
    """
    db_prediction = DBpredictions(**prediction, prediction_id=generate_id())
    session.add(db_prediction)
    session.commit()
    session.refresh(db_prediction)
    return db_prediction

generate_id()

Generate a unique string ID.

Returns:
  • str –

    A randomly generated unique string ID.
Source code in api/database.py 
78
79
80
81
82
83
84
85
86
87
def generate_id():
    """
    Generate a unique string ID.

    Returns:
        str: A randomly generated unique string ID.
    """
    length = 14
    characters = string.ascii_letters + string.digits
    return ''.join(random.choice(characters) for i in range(length))

get_db()

Dependency to get the database session.

Yields:
  • Session –

    SQLAlchemy database session.
Source code in api/database.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def get_db():
    """
    Dependency to get the database session.

    Yields:
        Session: SQLAlchemy database session.
    """
    engine = connect_to_postgres()
    session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
    Base.metadata.create_all(bind=engine)
    database = session_local()
    try:
        yield database
    finally:
        database.close()