92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
from datetime import datetime
|
|
from database import get_connection, is_postgres
|
|
|
|
class Destinations:
|
|
def __init__(self):
|
|
# Parameter style: Postgres uses %s, SQLite uses ?
|
|
self.ph = "%s" if is_postgres() else "?"
|
|
|
|
def destination_to_dict(self, row):
|
|
destination = {
|
|
"id": row[0],
|
|
"user_id": row[1],
|
|
"name": row[2],
|
|
"address": row[3],
|
|
"latitude": row[4],
|
|
"longitude": row[5],
|
|
"created_at": row[6]
|
|
}
|
|
return destination
|
|
|
|
def create(self, user_id, name, address):
|
|
created_at = datetime.now().isoformat()
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""
|
|
INSERT INTO destinations (user_id, name, address, created_at)
|
|
VALUES ({self.ph}, {self.ph}, {self.ph}, {self.ph})
|
|
""",
|
|
(user_id, name, address, created_at),
|
|
)
|
|
if hasattr(conn, "commit"):
|
|
conn.commit()
|
|
return cur.lastrowid if hasattr(cur, "lastrowid") else None
|
|
|
|
def update(self, id, user_id, name, address):
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""
|
|
UPDATE destinations
|
|
SET user_id = {self.ph}, name = {self.ph}, address = {self.ph}
|
|
WHERE id = {self.ph}
|
|
""",
|
|
(user_id, name, address, id),
|
|
)
|
|
if hasattr(conn, "commit"):
|
|
conn.commit()
|
|
|
|
def add_gps_coordinates(self, id, latitude, longitude):
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"""
|
|
UPDATE destinations
|
|
SET latitude = {self.ph}, longitude = {self.ph}
|
|
WHERE id = {self.ph}
|
|
""",
|
|
(latitude, longitude, id),
|
|
)
|
|
if hasattr(conn, "commit"):
|
|
conn.commit()
|
|
|
|
def delete(self, id):
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"DELETE FROM destinations WHERE id = {self.ph}",
|
|
(id,),
|
|
)
|
|
if hasattr(conn, "commit"):
|
|
conn.commit()
|
|
|
|
def get_all_by_user(self, user_id):
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"SELECT * FROM destinations WHERE user_id = {self.ph} ORDER BY created_at DESC",
|
|
(user_id,),
|
|
)
|
|
rows = cur.fetchall()
|
|
return [self.destination_to_dict(row) for row in rows]
|
|
|
|
def get_by_id(self, id):
|
|
with get_connection() as conn:
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
f"SELECT * FROM destinations WHERE id = {self.ph}",
|
|
(id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return self.destination_to_dict(row) if row else None |