app.py
unknown
python
2 years ago
8.9 kB
33
Indexable
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()
Editor is loading...