Select Git revision
UserTest.php
-
Laurent Destailleur authoredLaurent Destailleur authored
StudentsDatabase.py 4.64 KiB
from typing import Union
from _mysql_connector import MySQLInterfaceError
from mysql.connector import errors
from sqlalchemy import create_engine, Column, Integer, String, Date, exc, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session #, relationship
def semester_code_is_well_formed(current_semester_code: str) -> bool:
is_well_formed = (len(current_semester_code) == 4 and current_semester_code.isdigit()
and current_semester_code[0] == '1' and current_semester_code[-1] in {'1', '5', '8'})
# we'll assume we don't want to go back to the 20th century
# we'll assume semesters only start in January, May, or August
if not is_well_formed:
print(f'{current_semester_code} is not a valid semester code. '
f'See https://registrar.unl.edu/academic-standards/policies/year-term-identifier/')
return is_well_formed
Base = declarative_base()
class Student(Base):
__tablename__ = 'students'
id = Column(Integer, primary_key=True)
semester_code = Column(String(4))
nuid = Column(String(11), unique=True)
first_name = Column(String(50))
middle_name = Column(String(255))
last_name = Column(String(50))
rest_name = Column(String(15))
enrolled = Column(Integer)
college = Column(String(15))
major = Column(String(15))
student_year = Column(String(10))
gpa = Column(String(15))
raiks_student = Column(Integer)
email = Column(String(255))
unl_uid = Column(String(255))
private = Column(Integer)
dropped = Column(Integer)
insert_date = Column(Date)
# Relationship to Enrollments
# enrollments = relationship("Enrollment", back_populates="student")
class Enrollment(Base):
__tablename__ = 'enrollments'
id = Column(Integer, primary_key=True)
semester_code = Column(String(4))
nuid = Column(String(11), ForeignKey('students.nuid'))
course = Column(String(11))
section = Column(String(11))
grade = Column(String(5))
override = Column(String(25))
dropped = Column(Integer)
insert_date = Column(Date)
# Relationship to Student
# student = relationship("Student", back_populates="enrollments")
# Relationship to CourseSchedule
# noinspection PyTypeChecker
# course_schedule = relationship("CourseSchedule", back_populates="enrollments", foreign_keys=[course, section])
class CourseSchedule(Base):
__tablename__ = 'course_schedules'
id = Column(Integer, primary_key=True)
academic_session_id = Column(Integer)
semester_code = Column(String(4))
call_number = Column(String(11))
course = Column(String(11))
section = Column(String(15))
title = Column(String(50))
topic = Column(String(255))
department = Column(String(15))
credits = Column(String(4))
enrollment = Column(Integer)
instructor = Column(String(50))
instructor_nuid = Column(String(12))
activity = Column(String(5))
building = Column(String(15))
room = Column(String(10))
days = Column(String(7))
start_time = Column(String(10))
end_time = Column(String(10))
active = Column(Integer)
hidden = Column(Integer)
override = Column(Integer)
insert_date = Column(Date)
# Relationship to Enrollments
# enrollments = relationship("Enrollment", back_populates="course_schedule")
class StudentsDatabase(object):
@staticmethod
def connect(username: str, password: str,
host: str = 'cse-apps.unl.edu', port: int = 3306, database: str = 'students') -> Session:
url = StudentsDatabase.construct_mysql_url(host, port, database, username, password)
try:
students_database = StudentsDatabase(url)
session = students_database.create_session()
except exc.ProgrammingError | errors.ProgrammingError | MySQLInterfaceError as error:
# we don't seem to actually be able to catch the exception!
print('Could not connect to the database.')
print(error)
exit(1)
return session
@staticmethod
def construct_mysql_url(authority: str, port: Union[int, str], database: str, username: str, password: str) -> str:
return f'mysql+mysqlconnector://{username}:{password}@{authority}:{port}/{database}'
@staticmethod
def construct_in_memory_url() -> str:
return 'sqlite:///'
def __init__(self, url):
self.engine = create_engine(url)
self.Session = sessionmaker()
self.Session.configure(bind=self.engine)
def ensure_tables_exist(self) -> None:
Base.metadata.create_all(self.engine)
def create_session(self) -> Session:
return self.Session()