-
Notifications
You must be signed in to change notification settings - Fork 0
/
example.py
76 lines (56 loc) · 1.92 KB
/
example.py
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
65
66
67
68
69
70
71
72
73
74
75
76
from dataclasses import dataclass, field
from typing import Iterable, Generator, Tuple
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker
from py_frm import Base, sqlalchemy_model, to_sqlalchemy_query, model_for
@sqlalchemy_model(table="students")
@dataclass
class Student:
student_id: int = field(metadata={"primary_key": True})
name: str
@sqlalchemy_model(table="courses")
@dataclass
class Course:
course_id: int = field(metadata={"primary_key": True})
title: str
student_id: int = field(metadata={"foreign_key": ("students", "student_id")})
def get_student_courses(
students: Iterable[Student], courses: Iterable[Course]
) -> Generator[Tuple[str, str], None, None]:
return (
(s.name, c.title)
for s in students
for c in courses
if s.student_id == c.student_id
)
def populate_db(session):
StudentModel = model_for("students")
CourseModel = model_for("courses")
students = [
StudentModel(student_id=1, name="Alice"),
StudentModel(student_id=2, name="Bob"),
StudentModel(student_id=3, name="Charlie"),
]
session.add_all(students)
courses = [
CourseModel(course_id=1, title="Mathematics", student_id=1),
CourseModel(course_id=2, title="Physics", student_id=2),
CourseModel(course_id=3, title="Chemistry", student_id=1),
]
session.add_all(courses)
session.commit()
def main():
engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(engine)
inspector = inspect(engine)
Session = sessionmaker(bind=engine)
session = Session()
tables = inspector.get_table_names()
print("Tables in the database:", tables)
populate_db(session)
query = to_sqlalchemy_query(get_student_courses, session)
print(f"SQL: {query.statement}")
for result in query.all():
print(result)
if __name__ == "__main__":
main()