-
Notifications
You must be signed in to change notification settings - Fork 285
Expand file tree
/
Copy pathdatabase.py
More file actions
64 lines (51 loc) · 2.27 KB
/
database.py
File metadata and controls
64 lines (51 loc) · 2.27 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
# Database setup and ORM
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DateTime, Boolean
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
import bcrypt
import datetime # Using standard datetime for now, will integrate Jalali later
# TODO: Consider PostgreSQL for larger scale
DATABASE_URL = "sqlite:///fleet_management.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# --- User Model and Authentication ---
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
role = Column(String, nullable=False) # "admin", "operator"
is_active = Column(Boolean, default=True)
def set_password(self, password):
self.hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
def check_password(self, password):
return bcrypt.checkpw(password.encode('utf-8'), self.hashed_password.encode('utf-8'))
# --- Import actual models from their modules ---
# These imports are crucial for SQLAlchemy to recognize the tables.
from vehicle_management.models import Vehicle
from driver_management.models import Driver
from shift_planning.models import Shift, ShiftAssignment # Ensure all relevant models are imported
from mission_management.models import Mission
# (User model is already defined above)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def create_tables():
Base.metadata.create_all(bind=engine)
if __name__ == "__main__":
create_tables()
print("Database tables created successfully.")
# Example: Create a default admin user
db_session = SessionLocal()
admin_user = db_session.query(User).filter(User.username == "admin").first()
if not admin_user:
admin_user = User(username="admin", role="admin")
admin_user.set_password("admin123") # Default password, change in production
db_session.add(admin_user)
db_session.commit()
print("Default admin user created.")
db_session.close()