87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
from datetime import datetime
|
|
|
|
from database import get_connection, is_postgres
|
|
|
|
class Transporters:
|
|
def __init__(self):
|
|
# Parameter placeholder depending on backend
|
|
self.ph = "%s" if is_postgres() else "?"
|
|
|
|
def transporter_to_dict(self, row):
|
|
return {
|
|
"id": row[0],
|
|
"user_id": row[1],
|
|
"name": row[2],
|
|
"address": row[3],
|
|
"register_number": row[4],
|
|
"contact_person": row[5],
|
|
"phone": row[6],
|
|
"email": row[7],
|
|
"vat": row[8],
|
|
"created_at": row[9]
|
|
}
|
|
|
|
def create_transporter(self, name, address, register_number, contact_person, phone, email, vat, user_id):
|
|
created_at = datetime.now().isoformat()
|
|
returning = " RETURNING id" if is_postgres() else ""
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
f"""
|
|
INSERT INTO transporters
|
|
(name, address, register_number, contact_person, phone, email, vat, created_at, user_id)
|
|
VALUES ({self.ph}, {self.ph}, {self.ph}, {self.ph}, {self.ph}, {self.ph}, {self.ph}, {self.ph}, {self.ph}){returning}
|
|
""",
|
|
(name, address, register_number, contact_person, phone, email, vat, created_at, user_id),
|
|
)
|
|
transporter_id = cursor.fetchone()[0] if is_postgres() else cursor.lastrowid
|
|
if hasattr(conn, "commit"):
|
|
conn.commit()
|
|
return transporter_id
|
|
|
|
def get_all_transporters_by_user(self, user_id):
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
f"SELECT * FROM transporters WHERE user_id = {self.ph} ORDER BY created_at DESC",
|
|
(user_id,),
|
|
)
|
|
rows = cursor.fetchall()
|
|
return [self.transporter_to_dict(row) for row in rows]
|
|
|
|
def get_transporter_by_id(self, transporter_id):
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
f"SELECT * FROM transporters WHERE id = {self.ph}",
|
|
(transporter_id,),
|
|
)
|
|
row = cursor.fetchone()
|
|
return self.transporter_to_dict(row) if row else None
|
|
|
|
def update_transporter(self, transporter_id, name, address, register_number, contact_person, phone, email, vat):
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
f"""
|
|
UPDATE transporters SET
|
|
name = {self.ph},
|
|
address = {self.ph},
|
|
register_number = {self.ph},
|
|
contact_person = {self.ph},
|
|
phone = {self.ph},
|
|
email = {self.ph},
|
|
vat = {self.ph}
|
|
WHERE id = {self.ph}
|
|
""",
|
|
(name, address, register_number, contact_person, phone, email, vat, transporter_id),
|
|
)
|
|
if hasattr(conn, "commit"):
|
|
conn.commit()
|
|
|
|
def delete_transporter(self, transporter_id):
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(f"DELETE FROM transporters WHERE id = {self.ph}", (transporter_id,))
|
|
if hasattr(conn, "commit"):
|
|
conn.commit() |