backend.im/multimodalticketingsystem-rvf3wy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Multimodal Ticketing System with FastAPI and SQLite

Browse Source 
This commit includes:
- Project structure and FastAPI setup
- SQLAlchemy models for users, vehicles, schedules, and tickets
- Alembic migrations
- User authentication and management
- Vehicle and schedule management
- Ticket purchase and cancellation with time restrictions
- Comprehensive API documentation
This commit is contained in:
Automated Action 2025-06-17 11:08:42 +00:00
parent f65c8c2a49
commit 9e56bda916
31 changed files with 1908 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
README.md
View File

@ -1,3 +1,123 @@
# FastAPI Application
# Multimodal Ticketing System
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A backend system built with FastAPI that allows users to purchase tickets for different transportation modes (cars, buses, trains).
## Features
- User registration and authentication
- User profile management
- Vehicle management (cars, buses, trains)
- Schedule management for different vehicles
- Ticket purchasing with time restrictions
- Ticket cancellation with time restrictions
- View active tickets and ticket history
## Requirements
- Python 3.8+
- SQLite database
## Environment Variables
The application uses the following environment variables:
| Variable | Description | Default |
|----------|-------------|---------|
| SECRET_KEY | JWT secret key | "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" |
| ACCESS_TOKEN_EXPIRE_MINUTES | Token expiration time in minutes | 30 |
## Installation
1. Clone the repository:
```bash
git clone <repository-url>
cd multimodalticketingsystem
```
2. Create a virtual environment and activate it:
```bash
python -m venv venv
source venv/bin/activate # On Windows, use `venv\Scripts\activate`
```
3. Install the required dependencies:
```bash
pip install -r requirements.txt
```
4. Run the database migrations:
```bash
alembic upgrade head
```
5. Start the application:
```bash
uvicorn main:app --reload
```
The API will be available at http://localhost:8000.
## API Documentation
The API documentation is available at:
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
## API Endpoints
### Authentication
- `POST /api/v1/auth/register` - Register a new user
- `POST /api/v1/auth/login` - Login and get access token
### Users
- `GET /api/v1/users/me` - Get current user profile
- `PUT /api/v1/users/me` - Update current user profile
### Vehicles
- `GET /api/v1/vehicles` - List all vehicles
- `POST /api/v1/vehicles` - Create a new vehicle
- `GET /api/v1/vehicles/{vehicle_id}` - Get vehicle details
- `PUT /api/v1/vehicles/{vehicle_id}` - Update a vehicle
- `DELETE /api/v1/vehicles/{vehicle_id}` - Delete a vehicle (soft delete)
### Schedules
- `GET /api/v1/vehicles/schedules` - List all schedules
- `POST /api/v1/vehicles/schedules` - Create a new schedule
- `GET /api/v1/vehicles/schedules/{schedule_id}` - Get schedule details
- `PUT /api/v1/vehicles/schedules/{schedule_id}` - Update a schedule
- `DELETE /api/v1/vehicles/schedules/{schedule_id}` - Delete a schedule (soft delete)
### Tickets
- `POST /api/v1/tickets` - Purchase a ticket
- `GET /api/v1/tickets` - List all tickets for current user
- `GET /api/v1/tickets/active` - List active tickets for current user
- `GET /api/v1/tickets/history` - List ticket history for current user
- `GET /api/v1/tickets/{ticket_id}` - Get ticket details by ID
- `GET /api/v1/tickets/by-ticket-number/{ticket_number}` - Get ticket details by ticket number
- `PUT /api/v1/tickets/{ticket_id}/cancel` - Cancel a ticket
## Business Rules
- Tickets cannot be purchased less than 10 minutes before departure time
- Tickets cannot be cancelled less than 3 minutes before departure time
- Train tickets include a seat number, while car and bus tickets do not
- When a ticket is cancelled, the seat becomes available for purchase again
## Database Structure
The system uses the following database models:
- **Users**: Store user information
- **Vehicles**: Store different types of vehicles (cars, buses, trains)
- **Schedules**: Store departure and arrival information for vehicles
- **Tickets**: Store ticket information linked to users and schedules

86
alembic.ini Normal file
View File

@ -0,0 +1,86 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# timezone to use when rendering the date
# within the migration file as well as the filename.
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; this defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path
# version_locations = %(here)s/bar %(here)s/bat migrations/versions
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# SQLite URL example with absolute path
sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks=black
# black.type=console_scripts
# black.entrypoint=black
# black.options=-l 79
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

10
app/api/v1/api.py Normal file
View File

@ -0,0 +1,10 @@
from fastapi import APIRouter
from app.api.v1.endpoints import auth, tickets, users, vehicles
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(vehicles.router, prefix="/vehicles", tags=["vehicles"])
api_router.include_router(tickets.router, prefix="/tickets", tags=["tickets"])

79
app/api/v1/endpoints/auth.py Normal file
View File

@ -0,0 +1,79 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.core.auth import authenticate_user
from app.core.config import settings
from app.core.security import create_access_token, get_password_hash
from app.db.session import get_db
from app.models.user import User
from app.schemas.token import Token
from app.schemas.user import User as UserSchema
from app.schemas.user import UserCreate
router = APIRouter()
@router.post("/login", response_model=Token)
def login_access_token(
db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests.
"""
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/register", response_model=UserSchema, status_code=status.HTTP_201_CREATED)
def register_user(
*,
db: Session = Depends(get_db),
user_in: UserCreate,
) -> Any:
"""
Register a new user.
"""
# Check if username already exists
user = db.query(User).filter(User.username == user_in.username).first()
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered",
)
# Check if email already exists
user = db.query(User).filter(User.email == user_in.email).first()
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered",
)
# Create new user
user = User(
username=user_in.username,
email=user_in.email,
hashed_password=get_password_hash(user_in.password),
is_active=True,
)
db.add(user)
db.commit()
db.refresh(user)
return user

238
app/api/v1/endpoints/tickets.py Normal file
View File

@ -0,0 +1,238 @@
from datetime import datetime
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.auth import get_current_active_user
from app.db.session import get_db
from app.models.ticket import Ticket, TicketStatus
from app.models.user import User
from app.models.vehicle import Schedule, Vehicle, VehicleType
from app.schemas.ticket import (
Ticket as TicketSchema,
)
from app.schemas.ticket import (
TicketCreate,
)
from app.services.ticket_service import (
assign_seat_number,
generate_ticket_number,
validate_cancellation_time,
validate_purchase_time,
)
router = APIRouter()
@router.post("/", response_model=TicketSchema, status_code=status.HTTP_201_CREATED)
def create_ticket(
*,
db: Session = Depends(get_db),
ticket_in: TicketCreate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Purchase a ticket.
"""
# Check if schedule exists and is active
schedule = db.query(Schedule).join(Vehicle).filter(
Schedule.id == ticket_in.schedule_id,
Schedule.is_active == True,
Vehicle.is_active == True
).first()
if not schedule:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Schedule not found",
)
# Check if there are available seats
if schedule.available_seats <= 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No available seats for this schedule",
)
# Validate purchase time (at least 10 minutes before departure)
validate_purchase_time(schedule)
# For trains, assign a seat number
vehicle_type = schedule.vehicle.vehicle_type
if vehicle_type == VehicleType.TRAIN:
seat_number = assign_seat_number(db, schedule, vehicle_type)
else:
seat_number = ticket_in.seat_number
# Create ticket
ticket = Ticket(
user_id=current_user.id,
schedule_id=ticket_in.schedule_id,
seat_number=seat_number,
purchase_time=datetime.utcnow(),
status=TicketStatus.ACTIVE,
is_active=True,
ticket_number=generate_ticket_number(),
)
# Update available seats in schedule
schedule.available_seats -= 1
db.add(ticket)
db.add(schedule)
db.commit()
db.refresh(ticket)
return ticket
@router.get("/", response_model=List[TicketSchema])
def list_tickets(
*,
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
status: Optional[TicketStatus] = None,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve all tickets for the current user.
"""
query = db.query(Ticket).filter(
Ticket.user_id == current_user.id,
Ticket.is_active == True
)
if status:
query = query.filter(Ticket.status == status)
tickets = query.offset(skip).limit(limit).all()
return tickets
@router.get("/active", response_model=List[TicketSchema])
def list_active_tickets(
*,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve all active tickets for the current user.
"""
tickets = db.query(Ticket).filter(
Ticket.user_id == current_user.id,
Ticket.status == TicketStatus.ACTIVE,
Ticket.is_active == True
).all()
return tickets
@router.get("/history", response_model=List[TicketSchema])
def list_ticket_history(
*,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve ticket history (used, cancelled, expired) for the current user.
"""
tickets = db.query(Ticket).filter(
Ticket.user_id == current_user.id,
Ticket.status != TicketStatus.ACTIVE,
Ticket.is_active == True
).all()
return tickets
@router.get("/{ticket_id}", response_model=TicketSchema)
def get_ticket(
*,
db: Session = Depends(get_db),
ticket_id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Get ticket by ID.
"""
ticket = db.query(Ticket).filter(
Ticket.id == ticket_id,
Ticket.user_id == current_user.id,
Ticket.is_active == True
).first()
if not ticket:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Ticket not found",
)
return ticket
@router.get("/by-ticket-number/{ticket_number}", response_model=TicketSchema)
def get_ticket_by_number(
*,
db: Session = Depends(get_db),
ticket_number: str,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Get ticket by ticket number.
"""
ticket = db.query(Ticket).filter(
Ticket.ticket_number == ticket_number,
Ticket.user_id == current_user.id,
Ticket.is_active == True
).first()
if not ticket:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Ticket not found",
)
return ticket
@router.put("/{ticket_id}/cancel", response_model=TicketSchema)
def cancel_ticket(
*,
db: Session = Depends(get_db),
ticket_id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Cancel a ticket.
"""
ticket = db.query(Ticket).filter(
Ticket.id == ticket_id,
Ticket.user_id == current_user.id,
Ticket.is_active == True,
Ticket.status == TicketStatus.ACTIVE
).first()
if not ticket:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Active ticket not found",
)
# Validate cancellation time (at least 3 minutes before departure)
validate_cancellation_time(ticket)
# Update ticket status
ticket.status = TicketStatus.CANCELLED
# Increment available seats in schedule
schedule = db.query(Schedule).filter(Schedule.id == ticket.schedule_id).first()
schedule.available_seats += 1
db.add(ticket)
db.add(schedule)
db.commit()
db.refresh(ticket)
return ticket

67
app/api/v1/endpoints/users.py Normal file
View File

@ -0,0 +1,67 @@
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core.auth import get_current_active_user
from app.core.security import get_password_hash
from app.db.session import get_db
from app.models.user import User
from app.schemas.user import User as UserSchema
from app.schemas.user import UserUpdate
router = APIRouter()
@router.get("/me", response_model=UserSchema)
def read_user_me(
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Get current user.
"""
return current_user
@router.put("/me", response_model=UserSchema)
def update_user_me(
*,
db: Session = Depends(get_db),
user_in: UserUpdate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Update own user.
"""
# Check if username is being updated and if it's already taken
if user_in.username and user_in.username != current_user.username:
user = db.query(User).filter(User.username == user_in.username).first()
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Username already registered",
)
# Check if email is being updated and if it's already taken
if user_in.email and user_in.email != current_user.email:
user = db.query(User).filter(User.email == user_in.email).first()
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email already registered",
)
# Update user details
user_data = user_in.dict(exclude_unset=True)
if user_in.password:
user_data["hashed_password"] = get_password_hash(user_in.password)
del user_data["password"]
for key, value in user_data.items():
setattr(current_user, key, value)
db.add(current_user)
db.commit()
db.refresh(current_user)
return current_user

353
app/api/v1/endpoints/vehicles.py Normal file
View File

@ -0,0 +1,353 @@
from datetime import datetime
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import or_
from sqlalchemy.orm import Session
from app.core.auth import get_current_active_user
from app.db.session import get_db
from app.models.user import User
from app.models.vehicle import Schedule, Vehicle, VehicleType
from app.schemas.vehicle import (
Schedule as ScheduleSchema,
)
from app.schemas.vehicle import (
ScheduleCreate,
ScheduleUpdate,
VehicleCreate,
VehicleUpdate,
)
from app.schemas.vehicle import (
Vehicle as VehicleSchema,
)
router = APIRouter()
# Vehicle endpoints
@router.get("/", response_model=List[VehicleSchema])
def list_vehicles(
*,
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
vehicle_type: Optional[VehicleType] = None,
search: Optional[str] = None,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve vehicles with optional filtering.
"""
query = db.query(Vehicle).filter(Vehicle.is_active == True)
if vehicle_type:
query = query.filter(Vehicle.vehicle_type == vehicle_type)
if search:
query = query.filter(
or_(
Vehicle.vehicle_number.contains(search),
)
)
vehicles = query.offset(skip).limit(limit).all()
return vehicles
@router.post("/", response_model=VehicleSchema, status_code=status.HTTP_201_CREATED)
def create_vehicle(
*,
db: Session = Depends(get_db),
vehicle_in: VehicleCreate,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Create new vehicle.
"""
# Check if vehicle with same number already exists
db_vehicle = db.query(Vehicle).filter(Vehicle.vehicle_number == vehicle_in.vehicle_number).first()
if db_vehicle:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Vehicle with this number already exists",
)
vehicle = Vehicle(
vehicle_number=vehicle_in.vehicle_number,
vehicle_type=vehicle_in.vehicle_type,
capacity=vehicle_in.capacity,
is_active=True,
)
db.add(vehicle)
db.commit()
db.refresh(vehicle)
return vehicle
@router.get("/{vehicle_id}", response_model=VehicleSchema)
def get_vehicle(
*,
db: Session = Depends(get_db),
vehicle_id: int,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Get vehicle by ID.
"""
vehicle = db.query(Vehicle).filter(Vehicle.id == vehicle_id, Vehicle.is_active == True).first()
if not vehicle:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vehicle not found"
)
return vehicle
@router.put("/{vehicle_id}", response_model=VehicleSchema)
def update_vehicle(
*,
db: Session = Depends(get_db),
vehicle_id: int,
vehicle_in: VehicleUpdate,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Update a vehicle.
"""
vehicle = db.query(Vehicle).filter(Vehicle.id == vehicle_id, Vehicle.is_active == True).first()
if not vehicle:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vehicle not found"
)
# Check if vehicle number is being updated and if it's already taken
if vehicle_in.vehicle_number and vehicle_in.vehicle_number != vehicle.vehicle_number:
db_vehicle = db.query(Vehicle).filter(Vehicle.vehicle_number == vehicle_in.vehicle_number).first()
if db_vehicle:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Vehicle with this number already exists",
)
vehicle_data = vehicle_in.dict(exclude_unset=True)
for key, value in vehicle_data.items():
setattr(vehicle, key, value)
db.add(vehicle)
db.commit()
db.refresh(vehicle)
return vehicle
@router.delete("/{vehicle_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_vehicle(
*,
db: Session = Depends(get_db),
vehicle_id: int,
_: User = Depends(get_current_active_user),
) -> None:
"""
Delete a vehicle (soft delete).
"""
vehicle = db.query(Vehicle).filter(Vehicle.id == vehicle_id, Vehicle.is_active == True).first()
if not vehicle:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vehicle not found"
)
# Soft delete by setting is_active to False
vehicle.is_active = False
db.add(vehicle)
db.commit()
return None
# Schedule endpoints
@router.get("/schedules", response_model=List[ScheduleSchema])
def list_schedules(
*,
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
vehicle_id: Optional[int] = None,
vehicle_type: Optional[VehicleType] = None,
departure_from: Optional[datetime] = None,
departure_to: Optional[datetime] = None,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve schedules with optional filtering.
"""
query = db.query(Schedule).join(Vehicle).filter(
Schedule.is_active == True,
Vehicle.is_active == True
)
if vehicle_id:
query = query.filter(Schedule.vehicle_id == vehicle_id)
if vehicle_type:
query = query.filter(Vehicle.vehicle_type == vehicle_type)
if departure_from:
query = query.filter(Schedule.departure_time >= departure_from)
if departure_to:
query = query.filter(Schedule.departure_time <= departure_to)
schedules = query.order_by(Schedule.departure_time).offset(skip).limit(limit).all()
return schedules
@router.post("/schedules", response_model=ScheduleSchema, status_code=status.HTTP_201_CREATED)
def create_schedule(
*,
db: Session = Depends(get_db),
schedule_in: ScheduleCreate,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Create new schedule.
"""
# Check if vehicle exists
vehicle = db.query(Vehicle).filter(
Vehicle.id == schedule_in.vehicle_id,
Vehicle.is_active == True
).first()
if not vehicle:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Vehicle not found",
)
# Validate that arrival time is after departure time
if schedule_in.arrival_time <= schedule_in.departure_time:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Arrival time must be after departure time",
)
# Validate that available seats doesn't exceed vehicle capacity
if schedule_in.available_seats > vehicle.capacity:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Available seats cannot exceed vehicle capacity ({vehicle.capacity})",
)
schedule = Schedule(
vehicle_id=schedule_in.vehicle_id,
departure_location=schedule_in.departure_location,
arrival_location=schedule_in.arrival_location,
departure_time=schedule_in.departure_time,
arrival_time=schedule_in.arrival_time,
available_seats=schedule_in.available_seats,
is_active=True,
)
db.add(schedule)
db.commit()
db.refresh(schedule)
return schedule
@router.get("/schedules/{schedule_id}", response_model=ScheduleSchema)
def get_schedule(
*,
db: Session = Depends(get_db),
schedule_id: int,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Get schedule by ID.
"""
schedule = db.query(Schedule).filter(
Schedule.id == schedule_id,
Schedule.is_active == True
).first()
if not schedule:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Schedule not found"
)
return schedule
@router.put("/schedules/{schedule_id}", response_model=ScheduleSchema)
def update_schedule(
*,
db: Session = Depends(get_db),
schedule_id: int,
schedule_in: ScheduleUpdate,
_: User = Depends(get_current_active_user),
) -> Any:
"""
Update a schedule.
"""
schedule = db.query(Schedule).filter(
Schedule.id == schedule_id,
Schedule.is_active == True
).first()
if not schedule:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Schedule not found"
)
# If updating available seats, check that it doesn't exceed vehicle capacity
if schedule_in.available_seats is not None:
vehicle = db.query(Vehicle).filter(Vehicle.id == schedule.vehicle_id).first()
if schedule_in.available_seats > vehicle.capacity:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Available seats cannot exceed vehicle capacity ({vehicle.capacity})",
)
# If updating times, validate that arrival is after departure
new_departure = schedule_in.departure_time or schedule.departure_time
new_arrival = schedule_in.arrival_time or schedule.arrival_time
if new_arrival <= new_departure:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Arrival time must be after departure time",
)
schedule_data = schedule_in.dict(exclude_unset=True)
for key, value in schedule_data.items():
setattr(schedule, key, value)
db.add(schedule)
db.commit()
db.refresh(schedule)
return schedule
@router.delete("/schedules/{schedule_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_schedule(
*,
db: Session = Depends(get_db),
schedule_id: int,
_: User = Depends(get_current_active_user),
) -> None:
"""
Delete a schedule (soft delete).
"""
schedule = db.query(Schedule).filter(
Schedule.id == schedule_id,
Schedule.is_active == True
).first()
if not schedule:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Schedule not found"
)
# Soft delete by setting is_active to False
schedule.is_active = False
db.add(schedule)
db.commit()
return None

92
app/core/auth.py Normal file
View File

@ -0,0 +1,92 @@
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from app.core.config import settings
from app.core.security import verify_password
from app.db.session import get_db
from app.models.user import User
from app.schemas.token import TokenPayload
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
def authenticate_user(db: Session, username: str, password: str) -> Optional[User]:
"""
Verify username and password.
Args:
db: Database session
username: Username to verify
password: Password to verify
Returns:
User object if authentication successful, None otherwise
"""
user = db.query(User).filter(User.username == username).first()
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
"""
Get the current user based on JWT token.
Args:
db: Database session
token: JWT token
Returns:
Current user
Raises:
HTTPException: If token is invalid or user not found
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
user_id: int = int(payload.get("sub"))
if user_id is None:
raise credentials_exception
token_data = TokenPayload(sub=user_id)
except JWTError:
raise credentials_exception
user = db.query(User).filter(User.id == token_data.sub).first()
if user is None:
raise credentials_exception
if not user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return user
def get_current_active_user(current_user: User = Depends(get_current_user)) -> User:
"""
Get the current active user.
Args:
current_user: Current user
Returns:
Current active user
Raises:
HTTPException: If user is inactive
"""
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user

28
app/core/config.py Normal file
View File

@ -0,0 +1,28 @@
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)
# Base settings
PROJECT_NAME: str = "Multimodal Ticketing System"
PROJECT_DESCRIPTION: str = "A system for purchasing tickets for various transportation modes"
VERSION: str = "0.1.0"
API_V1_STR: str = "/api/v1"
# Security settings
SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" # CHANGE IN PRODUCTION!
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# Database settings
DB_DIR = Path("/app") / "storage" / "db"
SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
settings = Settings()
# Ensure database directory exists
settings.DB_DIR.mkdir(parents=True, exist_ok=True)

58
app/core/security.py Normal file
View File

@ -0,0 +1,58 @@
from datetime import datetime, timedelta
from typing import Any, Union
from jose import jwt
from passlib.context import CryptContext
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def create_access_token(subject: Union[str, Any], expires_delta: timedelta = None) -> str:
"""
Create JWT access token.
Args:
subject: The subject of the token, typically user_id
expires_delta: Optional expiration time delta
Returns:
The encoded JWT token
"""
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Verify that the plain password matches the hashed password.
Args:
plain_password: The password in plain text
hashed_password: The hashed password to compare against
Returns:
True if the password matches, False otherwise
"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""
Hash a password for storing.
Args:
password: The password to hash
Returns:
The hashed password
"""
return pwd_context.hash(password)

3
app/db/base.py Normal file
View File

@ -0,0 +1,3 @@
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

9
app/db/base_class.py Normal file
View File

@ -0,0 +1,9 @@
from app.db.base import Base
from app.models.ticket import Ticket
from app.models.user import User
from app.models.vehicle import Schedule, Vehicle
# This file imports all SQLAlchemy models to ensure they are registered with the Base metadata
# This file is used by Alembic for auto-generating migrations
__all__ = ["Base", "User", "Vehicle", "Schedule", "Ticket"]

26
app/db/session.py Normal file
View File

@ -0,0 +1,26 @@
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} # Only needed for SQLite
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
"""
Dependency for getting a database session.
Usage:
@app.get("/")
def read_item(db: Session = Depends(get_db)):
...
"""
db = SessionLocal()
try:
yield db
finally:
db.close()

13
app/models/__init__.py Normal file
View File

@ -0,0 +1,13 @@
from app.models.ticket import Ticket, TicketStatus
from app.models.user import User
from app.models.vehicle import Schedule, Vehicle, VehicleType
# This list should contain all models for Alembic to detect changes
__all__ = [
"User",
"Vehicle",
"VehicleType",
"Schedule",
"Ticket",
"TicketStatus",
]

34
app/models/ticket.py Normal file
View File

@ -0,0 +1,34 @@
from datetime import datetime
from enum import Enum as PyEnum
from sqlalchemy import Boolean, Column, DateTime, Enum, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class TicketStatus(str, PyEnum):
ACTIVE = "active"
USED = "used"
CANCELLED = "cancelled"
EXPIRED = "expired"
class Ticket(Base):
__tablename__ = "tickets"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"))
schedule_id = Column(Integer, ForeignKey("schedules.id"))
seat_number = Column(String, nullable=True) # Seat number is only required for trains
purchase_time = Column(DateTime, default=datetime.utcnow)
status = Column(Enum(TicketStatus), default=TicketStatus.ACTIVE)
is_active = Column(Boolean, default=True)
ticket_number = Column(String, unique=True, index=True)
# Relationships
user = relationship("User", back_populates="tickets")
schedule = relationship("Schedule", back_populates="tickets")
def __repr__(self):
return f"Ticket(id={self.id}, user={self.user_id}, schedule={self.schedule_id}, status={self.status})"

20
app/models/user.py Normal file
View File

@ -0,0 +1,20 @@
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
# Relationships
tickets = relationship("Ticket", back_populates="user")
def __repr__(self):
return f"User(id={self.id}, username={self.username}, email={self.email})"

48
app/models/vehicle.py Normal file
View File

@ -0,0 +1,48 @@
from enum import Enum as PyEnum
from sqlalchemy import Boolean, Column, DateTime, Enum, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class VehicleType(str, PyEnum):
CAR = "car"
BUS = "bus"
TRAIN = "train"
class Vehicle(Base):
__tablename__ = "vehicles"
id = Column(Integer, primary_key=True, index=True)
vehicle_number = Column(String, unique=True, index=True)
vehicle_type = Column(Enum(VehicleType), index=True)
capacity = Column(Integer)
is_active = Column(Boolean, default=True)
# Relationships
schedules = relationship("Schedule", back_populates="vehicle")
def __repr__(self):
return f"Vehicle(id={self.id}, number={self.vehicle_number}, type={self.vehicle_type})"
class Schedule(Base):
__tablename__ = "schedules"
id = Column(Integer, primary_key=True, index=True)
vehicle_id = Column(Integer, ForeignKey("vehicles.id"))
departure_location = Column(String)
arrival_location = Column(String)
departure_time = Column(DateTime, index=True)
arrival_time = Column(DateTime)
available_seats = Column(Integer)
is_active = Column(Boolean, default=True)
# Relationships
vehicle = relationship("Vehicle", back_populates="schedules")
tickets = relationship("Ticket", back_populates="schedule")
def __repr__(self):
return f"Schedule(id={self.id}, vehicle={self.vehicle_id}, departure={self.departure_time})"

18
app/schemas/__init__.py Normal file
View File

@ -0,0 +1,18 @@
from app.schemas.ticket import (
Ticket,
TicketCreate,
TicketUpdate,
TicketWithoutSchedule,
TicketWithUser,
)
from app.schemas.token import Token, TokenPayload
from app.schemas.user import User, UserCreate, UserInDB, UserUpdate
from app.schemas.vehicle import (
Schedule,
ScheduleCreate,
ScheduleUpdate,
ScheduleWithoutVehicle,
Vehicle,
VehicleCreate,
VehicleUpdate,
)

46
app/schemas/ticket.py Normal file
View File

@ -0,0 +1,46 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
from app.models.ticket import TicketStatus
from app.schemas.user import User
from app.schemas.vehicle import ScheduleWithoutVehicle
class TicketBase(BaseModel):
schedule_id: int
seat_number: Optional[str] = None # Optional for cars and buses
class TicketCreate(TicketBase):
pass
class TicketUpdate(BaseModel):
status: Optional[TicketStatus] = None
is_active: Optional[bool] = None
class TicketInDBBase(TicketBase):
id: int
user_id: int
purchase_time: datetime
status: TicketStatus
is_active: bool
ticket_number: str
class Config:
from_attributes = True
class Ticket(TicketInDBBase):
schedule: ScheduleWithoutVehicle
class TicketWithUser(Ticket):
user: User
class TicketWithoutSchedule(TicketInDBBase):
pass

12
app/schemas/token.py Normal file
View File

@ -0,0 +1,12 @@
from typing import Optional
from pydantic import BaseModel
class Token(BaseModel):
access_token: str
token_type: str
class TokenPayload(BaseModel):
sub: Optional[int] = None

34
app/schemas/user.py Normal file
View File

@ -0,0 +1,34 @@
from typing import Optional
from pydantic import BaseModel, EmailStr, Field
class UserBase(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
username: Optional[str] = Field(None, min_length=3, max_length=50)
email: Optional[EmailStr] = None
password: Optional[str] = Field(None, min_length=8)
class UserInDBBase(UserBase):
id: int
is_active: bool
class Config:
from_attributes = True
class User(UserInDBBase):
pass
class UserInDB(UserInDBBase):
hashed_password: str

73
app/schemas/vehicle.py Normal file
View File

@ -0,0 +1,73 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
from app.models.vehicle import VehicleType
class VehicleBase(BaseModel):
vehicle_number: str
vehicle_type: VehicleType
capacity: int = Field(..., gt=0)
class VehicleCreate(VehicleBase):
pass
class VehicleUpdate(BaseModel):
vehicle_number: Optional[str] = None
vehicle_type: Optional[VehicleType] = None
capacity: Optional[int] = Field(None, gt=0)
is_active: Optional[bool] = None
class VehicleInDBBase(VehicleBase):
id: int
is_active: bool
class Config:
from_attributes = True
class Vehicle(VehicleInDBBase):
pass
class ScheduleBase(BaseModel):
vehicle_id: int
departure_location: str
arrival_location: str
departure_time: datetime
arrival_time: datetime
available_seats: int = Field(..., ge=0)
class ScheduleCreate(ScheduleBase):
pass
class ScheduleUpdate(BaseModel):
departure_location: Optional[str] = None
arrival_location: Optional[str] = None
departure_time: Optional[datetime] = None
arrival_time: Optional[datetime] = None
available_seats: Optional[int] = Field(None, ge=0)
is_active: Optional[bool] = None
class ScheduleInDBBase(ScheduleBase):
id: int
is_active: bool
class Config:
from_attributes = True
class Schedule(ScheduleInDBBase):
vehicle: Vehicle
class ScheduleWithoutVehicle(ScheduleInDBBase):
pass

1
app/services/__init__.py Normal file
View File

@ -0,0 +1 @@
# This file is intentionally empty to mark the directory as a Python package

97
app/services/ticket_service.py Normal file
View File

@ -0,0 +1,97 @@
import uuid
from datetime import datetime, timedelta
from typing import Optional
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from app.models.ticket import Ticket, TicketStatus
from app.models.vehicle import Schedule, VehicleType
def generate_ticket_number() -> str:
"""Generate a unique ticket number."""
return str(uuid.uuid4())
def validate_purchase_time(schedule: Schedule) -> None:
"""
Validate that a ticket can be purchased based on departure time.
Args:
schedule: The schedule for which the ticket is being purchased
Raises:
HTTPException: If ticket cannot be purchased due to time restrictions
"""
now = datetime.utcnow()
# Check if it's less than 10 minutes to departure
if schedule.departure_time - now < timedelta(minutes=10):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot purchase ticket less than 10 minutes before departure",
)
def validate_cancellation_time(ticket: Ticket) -> None:
"""
Validate that a ticket can be cancelled based on departure time.
Args:
ticket: The ticket to be cancelled
Raises:
HTTPException: If ticket cannot be cancelled due to time restrictions
"""
now = datetime.utcnow()
# Check if it's less than 3 minutes to departure
if ticket.schedule.departure_time - now < timedelta(minutes=3):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot cancel ticket less than 3 minutes before departure",
)
# Check if ticket is already used, cancelled, or expired
if ticket.status != TicketStatus.ACTIVE:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Cannot cancel ticket with status {ticket.status}",
)
def assign_seat_number(db: Session, schedule: Schedule, vehicle_type: VehicleType) -> Optional[str]:
"""
Assign a seat number for a ticket if vehicle is a train.
For cars and buses, seat number is None.
Args:
db: Database session
schedule: The schedule for which the ticket is being purchased
vehicle_type: Type of vehicle
Returns:
Seat number string for trains, None for other vehicle types
"""
if vehicle_type != VehicleType.TRAIN:
return None
# For trains, find the next available seat number
# Get all occupied seats for this schedule
occupied_seats = db.query(Ticket.seat_number).filter(
Ticket.schedule_id == schedule.id,
Ticket.status == TicketStatus.ACTIVE,
Ticket.seat_number.isnot(None)
).all()
occupied_seat_numbers = [int(seat[0]) for seat in occupied_seats if seat[0] and seat[0].isdigit()]
# Find the first available seat
for seat_num in range(1, schedule.vehicle.capacity + 1):
if seat_num not in occupied_seat_numbers:
return str(seat_num)
# This should not happen if available_seats is managed correctly
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="No available seats",
)

65
main.py Normal file
View File

@ -0,0 +1,65 @@
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from app.api.v1.api import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
description=settings.PROJECT_DESCRIPTION,
version=settings.VERSION,
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
)
# Set up CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API router
app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/")
async def root():
"""
Root endpoint returning basic application information.
"""
return {
"title": settings.PROJECT_NAME,
"description": settings.PROJECT_DESCRIPTION,
"version": settings.VERSION,
"documentation": "/docs",
"health_check": "/health",
}
@app.get("/health", status_code=200)
async def health_check():
"""
Health check endpoint to verify the application is running.
"""
return {"status": "healthy"}
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title=settings.PROJECT_NAME,
version=settings.VERSION,
description=settings.PROJECT_DESCRIPTION,
routes=app.routes,
)
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

29
migrations/README Normal file
View File

@ -0,0 +1,29 @@
# Database Migrations
This directory contains database migrations for the Multimodal Ticketing System project.
## Migration Commands
To run migrations:
```bash
alembic upgrade head
```
To create a new migration:
```bash
alembic revision -m "description of changes"
```
To show current migration version:
```bash
alembic current
```
To show migration history:
```bash
alembic history
```

82
migrations/env.py Normal file
View File

@ -0,0 +1,82 @@
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.db.base import Base
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
render_as_batch=True, # Important for SQLite
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
is_sqlite = connection.dialect.name == 'sqlite'
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=is_sqlite, # Important for SQLite
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

24
migrations/script.py.mako Normal file
View File

@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

104
migrations/versions/1f3a9c5d40a1_initial_migration.py Normal file
View File

@ -0,0 +1,104 @@
"""Initial migration
Revision ID: 1f3a9c5d40a1
Revises:
Create Date: 2023-06-01 00:00:00.000000
"""
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision = '1f3a9c5d40a1'
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
# Create users table
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('username', sa.String(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
# Create vehicles table
op.create_table(
'vehicles',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('vehicle_number', sa.String(), nullable=False),
sa.Column('vehicle_type', sa.Enum('car', 'bus', 'train', name='vehicletype'), nullable=False),
sa.Column('capacity', sa.Integer(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_vehicles_id'), 'vehicles', ['id'], unique=False)
op.create_index(op.f('ix_vehicles_vehicle_number'), 'vehicles', ['vehicle_number'], unique=True)
op.create_index(op.f('ix_vehicles_vehicle_type'), 'vehicles', ['vehicle_type'], unique=False)
# Create schedules table
op.create_table(
'schedules',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('vehicle_id', sa.Integer(), nullable=False),
sa.Column('departure_location', sa.String(), nullable=False),
sa.Column('arrival_location', sa.String(), nullable=False),
sa.Column('departure_time', sa.DateTime(), nullable=False),
sa.Column('arrival_time', sa.DateTime(), nullable=False),
sa.Column('available_seats', sa.Integer(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
sa.ForeignKeyConstraint(['vehicle_id'], ['vehicles.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_schedules_departure_time'), 'schedules', ['departure_time'], unique=False)
op.create_index(op.f('ix_schedules_id'), 'schedules', ['id'], unique=False)
# Create tickets table
op.create_table(
'tickets',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('schedule_id', sa.Integer(), nullable=False),
sa.Column('seat_number', sa.String(), nullable=True),
sa.Column('purchase_time', sa.DateTime(), nullable=False),
sa.Column('status', sa.Enum('active', 'used', 'cancelled', 'expired', name='ticketstatus'), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
sa.Column('ticket_number', sa.String(), nullable=False),
sa.ForeignKeyConstraint(['schedule_id'], ['schedules.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_tickets_id'), 'tickets', ['id'], unique=False)
op.create_index(op.f('ix_tickets_ticket_number'), 'tickets', ['ticket_number'], unique=True)
def downgrade() -> None:
op.drop_index(op.f('ix_tickets_ticket_number'), table_name='tickets')
op.drop_index(op.f('ix_tickets_id'), table_name='tickets')
op.drop_table('tickets')
op.drop_index(op.f('ix_schedules_id'), table_name='schedules')
op.drop_index(op.f('ix_schedules_departure_time'), table_name='schedules')
op.drop_table('schedules')
op.drop_index(op.f('ix_vehicles_vehicle_type'), table_name='vehicles')
op.drop_index(op.f('ix_vehicles_vehicle_number'), table_name='vehicles')
op.drop_index(op.f('ix_vehicles_id'), table_name='vehicles')
op.drop_table('vehicles')
op.drop_index(op.f('ix_users_username'), table_name='users')
op.drop_index(op.f('ix_users_id'), table_name='users')
op.drop_index(op.f('ix_users_email'), table_name='users')
op.drop_table('users')
# Remove enum types (SQLite ignores this, but important for other databases)
op.execute('DROP TYPE IF EXISTS vehicletype;')
op.execute('DROP TYPE IF EXISTS ticketstatus;')

24
pyproject.toml Normal file
View File

@ -0,0 +1,24 @@
[tool.ruff]
line-length = 120
target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "I", "B", "C4", "SIM"]
# Ignore some linting errors that are common in FastAPI applications
ignore = ["B008", "E712", "B904"]
# Exclude a variety of commonly ignored directories.
exclude = [
".git",
".ruff_cache",
".venv",
"venv",
"__pycache__",
"migrations",
]
[tool.ruff.lint.isort]
known-third-party = ["fastapi", "pydantic", "sqlalchemy", "starlette", "jose", "passlib"]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401", "E402"]

13
requirements.txt Normal file
View File

@ -0,0 +1,13 @@
fastapi>=0.95.1
uvicorn>=0.22.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
sqlalchemy>=2.0.0
alembic>=1.11.0
python-jose[cryptography]>=3.3.0
passlib[bcrypt]>=1.7.4
python-multipart>=0.0.6
email-validator>=2.0.0
python-dotenv>=1.0.0
ruff>=0.0.262
pytest>=7.3.1