initial

src/db.py

@@ -0,0 +1,153 @@
+import psycopg2 as pg
+
+from .objects import User, Rights
+
+DB_AUTH_HARDCODED = {
+    "host": "127.0.0.1",
+    "port": 5432,
+    "dbname": "examdb",
+    "user": "postgres",
+    "password": "213k2010###"
+}
+
+def get_connection():
+    return pg.connect(**DB_AUTH_HARDCODED)
+
+def do_request(autocommit=False):
+    def upper_wrapper(func):
+        def wrapper(*args, **kwargs):
+            conn = get_connection()
+            cursor = conn.cursor()
+
+            try:
+                kwargs['cursor'] = cursor
+
+                result = func(*args, **kwargs)
+
+                if autocommit:
+                    conn.commit()
+
+                return result
+
+            except Exception as e:
+                print(f"Error: Can't request query to DB: {e}")
+                return None
+
+            finally:
+                cursor.close()
+                conn.close()
+
+        return  wrapper
+
+    return upper_wrapper
+
+@do_request()
+def auth(login: str, password: str, *, cursor) -> User | None:
+    cursor.execute("""
+        SELECT id, name, rights
+        FROM users
+        WHERE name = %s
+        AND password = %s;
+    """, (login, password))
+
+    user = cursor.fetchone()
+
+    if not user:
+        print("Warning: Login Forbidden: Can't find such user!")
+        return None
+
+    rights = None
+
+    match user[2]:
+        case "admin":
+            rights = Rights.ADMIN
+        case "customer":
+            rights = Rights.CLIENT
+        case "manager":
+            rights = Rights.MANAGER
+        case _:
+            return None
+
+    return User(
+        id=user[0],
+        name=user[1],
+        rights=rights
+    )
+
+@do_request()
+def get_free_numbers(*, cursor):
+    cursor.execute("""
+        SELECT *
+        FROM rooms
+        WHERE status = 'free';
+    """)
+
+    free = cursor.fetchall()
+
+    if not free:
+        return None
+
+    return free
+
+@do_request(autocommit=True)
+def update_number_status(number: str, checkin: str,
+                         checkout: str, user: User,
+                         *, cursor):
+    cursor.execute("""
+        SELECT password
+        FROM users
+        WHERE id = %s
+    """, (user.id,))
+
+    password = cursor.fetchone()
+
+    if not password:
+        return False
+
+    cursor.execute("""
+        SELECT id
+        FROM guests
+        WHERE name = %s
+        AND PHONE = %s
+    """, (user.name, password[0]))
+
+    guest = cursor.fetchone()
+
+    if not guest:
+        return False
+
+    cursor.execute("""
+        SELECT id
+        FROM rooms
+        WHERE number = %s;
+    """, (number,))
+
+    number_id = cursor.fetchone()
+
+    if not number_id:
+        return False
+
+    cursor.execute("""
+        SELECT guest, room
+        FROM bookings
+        WHERE guest = %s
+        AND room = %s;
+    """, (guest[0], number_id[0]))
+
+    request_exists = cursor.fetchone()
+
+    if request_exists:
+        return False
+
+    cursor.execute("""
+        INSERT INTO bookings(guest, room, checkin, checkout, status)
+        VALUES (%s, %s, %s, %s, 'active');
+    """, (guest[0], number_id[0], checkin, checkout))
+
+    cursor.execute("""
+        UPDATE rooms
+        SET status = 'booked'
+        WHERE number = %s;
+    """, (number,))
+
+    return True