app.py

mail@pastecode.io avatar
unknown
python
a year ago
8.9 kB
30
Indexable
Never
import socketserver
import http.server
import json

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import DeclarativeMeta
from sqlalchemy.orm.collections import InstrumentedList
from sqlalchemy.orm.state import InstanceState

from models import train, ticket, Station, Reservations, Base
from config import *

engine = create_engine("sqlite:///DataBase.db", echo=True)
Session = sessionmaker(bind=engine)
session = Session()

Base.metadata.create_all(engine)

# Set up the server host and port
host = 'localhost'
port = 8000

def commitData():
    session.query(train).delete()
    session.query(ticket).delete()
    session.query(Station).delete()

    Chur = Station(
        station_id="1", 
        name="Chur"
    )
    Filisur = Station(
        station_id="2", 
        name="Filisur"
    )
    nunin = Station(
        station_id="3",
        name="Ňunín"
    )
    Chalen = Station(
        station_id="4",
        name="Chálen" 
    )

    BerEx = train(
        train_id="1", 
        name="Bernina Express", 
        shortDescription="From glaciers to palms", 
        longDescription=longDescriptionA, 
        imageUrl="https://images.ctfassets.net/ugsztf9t14yz/2j6PkgcjkvJzn797rIEg3s/da13f093c91608f313f97857093b56c8/Bernina_Express_beim_Lago_Bianco.jpg?fm=avif&f=center&fit=fill&q=45&w=2522",
        rating="4.3",
        ridesPerDay="1",
        stations=[Chalen], 
        changeNum="0"
    )
    BerExLM = train(
        train_id="2", 
        name="Bernina Express - Last Minute", 
        shortDescription="Benefit from attractive prices", 
        longDescription=longDescriptionB, 
        imageUrl="https://images.ctfassets.net/ugsztf9t14yz/4XXQ9l3gwicxWQT68Mzmu5/51175e8492d42f663c30885263f68b9e/Oberalppass___Glacier_Express_AG_-_Stefan_Schlumpf.jpg?fm=avif&f=center&fit=fill&q=45&h=937&w=1667",
        rating="5.0",
        ridesPerDay="1",
        stations=[Chur, nunin], 
        changeNum="0"
    )
    BerHeEx = train(
        train_id="3", 
        name="Bernina Herb Express", 
        shortDescription="Journey into the Valposchiavo world of herbs", 
        longDescription=longDescriptionC, 
        imageUrl="url",
        rating="4.0",
        ridesPerDay="1",
        stations=[Chur, Filisur], 
        changeNum="0"
    )

    reservationA = Reservations(
        name=BerEx.name, 
        ReservationID="1", 
        fName="Jan", 
        lName="Novák", 
        personalID="123456789", 
        mail="novakjan@seznam.cz"
    )
    reservationB = Reservations(
        name=BerExLM.name, 
        ReservationID="2", 
        fName="Petr", 
        lName="Pán", 
        personalID="123456781", 
        mail="petrpan@email.cz"
    )

    BerExTicket = ticket(
        ticket_id="1", 
        train_id="1", 
        departureTime=Time1A, 
        arrivalTime=Time1B, 
        capacity="140", 
        classNumber="1", 
        price="169"
    )

    session.add(BerEx)
    session.add(BerExLM)
    session.add(BerHeEx)

    session.add(BerExTicket)
    
    session.add(Chur)
    session.add(Filisur)
    session.add(nunin)
    session.add(reservationA)
    session.add(reservationB)

commitData()

class CustomJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            # Convert the SQLAlchemy object to a dictionary
            return {c.name: getattr(obj, c.name) for c in obj.__table__.columns}
        elif isinstance(obj, InstrumentedList):
            # Convert the InstrumentedList to a list of dictionaries
            return [item.__dict__ for item in obj]
        elif isinstance(obj, InstanceState):
            # Convert the InstanceState object to a dictionary or any other JSON-serializable format
            return str(obj)  # Example: Convert to a string representation
        return super().default(obj)
    
class MyRequestHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == '/api/train':
            formatted_data = []
            DBdata = session.query(train).all()

            for result in DBdata:
                formatted_data.append({
                    'train_id': result.train_id,
                    'name': result.name,
                    'shortDescription': result.shortDescription,
                    'stations': result.stations,
                })

            json_data = json.dumps(formatted_data, cls=CustomJSONEncoder)
            
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()

            self.wfile.write(bytes(json_data, "utf-8"))  

        elif self.path == "/api/ticket":
            formatted_data_ticket = []
            DBdata_ticket = session.query(ticket).all()

            for resultTicket in DBdata_ticket:
                formatted_data_ticket.append({
                    'ticket_id': resultTicket.ticket_id,
                    'train_id': resultTicket.train_id,
                    'departureTime': resultTicket.departureTime.isoformat(),
                    'arrivalTime': resultTicket.arrivalTime.isoformat(),
                    'capacity': resultTicket.capacity,
                    'classNumber': resultTicket.classNumber,
                    'price': resultTicket.price,
                })
            
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()

            json_data_ticket = json.dumps(formatted_data_ticket)
            self.wfile.write(bytes(json_data_ticket, "utf-8"))

        elif self.path.startswith('/api/train/'):
            train_id = self.path.split('/')[-1]
            formatted_data_trainByID = []
            train_data = session.query(train).filter_by(train_id=train_id).first()

            if train_data:
                formatted_data_trainByID.append({
                    'train_id': train_data.train_id,
                    'name': train_data.name,
                    'longDescription': train_data.longDescription,
                    'shortDescription': train_data.shortDescription,
                    'imageUrl': train_data.imageUrl,
                    'rating': train_data.rating,
                    'ridesPerDay': train_data.ridesPerDay,
                    'stations': [station.__dict__ for station in train_data.stations],
                    'changeNum': train_data.changeNum,
                })

                json_data_trainByID = json.dumps(formatted_data_trainByID, cls=CustomJSONEncoder)

                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(bytes(json_data_trainByID, "utf-8"))
            else:
                self.send_response(404)  # Return a 404 status code
                self.end_headers()
                self.wfile.write(b'404 - Train Not Found')
        elif self.path == '/api/station':
            formatted_data_station = []
            DBdata_station = session.query(Station).all()

            for stationResult in DBdata_station:
                formatted_data_station.append({
                    'station_id': stationResult.station_id,
                    'name': stationResult.name,
                })

            json_data = json.dumps(formatted_data_station, cls=CustomJSONEncoder)
            
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()

            self.wfile.write(bytes(json_data, "utf-8"))
        elif self.path == '/api/reservations':
            formatted_data_reservations = []
            DBdata_reservations = session.query(Reservations).all()

            for reservationsResult in DBdata_reservations:
                formatted_data_reservations.append({
                    'personalID': reservationsResult.personalID,
                    'name': reservationsResult.name,
                    'fName': reservationsResult.fName,
                    'lName': reservationsResult.lName,
                    'reservationID': reservationsResult.ReservationID,
                    'mail': reservationsResult.mail,
                })

            json_data = json.dumps(formatted_data_reservations, cls=CustomJSONEncoder)
            
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()

            self.wfile.write(bytes(json_data, "utf-8"))
        else:
            self.send_response(404)  # Return a 404 status code 
            self.end_headers()

            self.wfile.write(b'404 - Not Found')  

PORT = 8000  # Specify the port number
Handler = MyRequestHandler

with socketserver.TCPServer(("", PORT), Handler) as httpd:
    print("Server started at port", PORT)
    httpd.serve_forever()