-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy pathexample.py
More file actions
114 lines (96 loc) · 4.49 KB
/
example.py
File metadata and controls
114 lines (96 loc) · 4.49 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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
from typing import List
from faker import Faker
from sqlalchemy import ForeignKey, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship
class Base(DeclarativeBase):
pass
class Customer(Base):
__tablename__ = "customer"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email_address: Mapped[str]
address: Mapped[str]
country_code: Mapped[str] = mapped_column(String(2))
# add a 1-to-1 relationship to CreditCard
credit_card: Mapped["CreditCard"] = relationship("CreditCard", uselist=False,
back_populates="customer")
# add a 1-to-many relationship to Order
orders: Mapped[List["Order"]] = relationship("Order", back_populates="customer")
def __repr__(self):
return f"<Customer(name={self.name!r})>"
class CreditCard(Base):
__tablename__ = "credit_card"
id: Mapped[int] = mapped_column(primary_key=True)
customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))
customer: Mapped[Customer] = relationship("Customer", back_populates="credit_card")
number: Mapped[str] = mapped_column(String(19))
def __repr__(self):
return f"<CreditCard(number={self.number!r})>"
# Add a Product with name, price, description, and category
class Product(Base):
__tablename__ = "product"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
price: Mapped[float]
description: Mapped[str]
category: Mapped[str] = mapped_column(String(100))
orders: Mapped[List["Order"]] = relationship("Order", back_populates="product")
def __repr__(self):
return f"<Product(name={self.name!r})>"
# Add an Order with a customer_id, product_id, and quantity
class Order(Base):
__tablename__ = "order"
id: Mapped[int] = mapped_column(primary_key=True)
customer_id: Mapped[int] = mapped_column(ForeignKey("customer.id"))
customer: Mapped[Customer] = relationship("Customer", back_populates="orders")
product_id: Mapped[int] = mapped_column(ForeignKey("product.id"))
product: Mapped[Product] = relationship("Product", back_populates="orders")
quantity: Mapped[int]
def __repr__(self):
return f"<Order(quantity={self.quantity!r})>"
# Set up a database connection and create tables
engine = create_engine("sqlite:///my_database.db", echo=True)
Base.metadata.create_all(engine)
fake = Faker(["en_US", "en_GB", "en_CA", "es_MX", "de_DE", "fr_FR", "ja_JP"])
with Session(engine) as session:
# insert 10 products
for i in range(10):
product = Product(name=f"Product {i}", price=9.99, description="ABC", category="XYZ")
session.add(product)
# create 100 customers
for i in range(100):
customer = Customer(name=fake.name(), email_address=fake.email(),
address=fake.address(), country_code=fake.country_code())
session.add(customer)
# create a credit card for each customer
credit_card = CreditCard(number=fake.credit_card_number(), customer=customer)
session.add(credit_card)
# insert random amount of orders of random product IDs using their credit card
for _ in range(fake.random_int(min=1, max=5)):
order = Order(customer=customer, product_id=fake.random_int(min=1, max=10),
quantity=fake.random_int(min=1, max=5))
session.add(order)
# commit the session to the database
session.commit()
# Query the database
# Get all customers
query = select(Customer)
results = session.execute(query).scalars().all()
print(results)
# Get all customers from the US
query = select(Customer).where(Customer.country_code == "US")
results = session.execute(query).scalars().all()
print(results)
# Select a list of countries grouped by country code
from sqlalchemy import func
query = select(Customer.country_code, func.count(Customer.country_code)).group_by(Customer.country_code)
results = session.execute(query).all()
print(results)
# Select customer name with credit card number
query = select(Customer.name, CreditCard.number).join(CreditCard)
results = session.execute(query).all()
print(results)
# Select customer name with their number of orders
query = select(Customer.id, func.count(Order.id)).join(Order).group_by(Customer.id)
results = session.execute(query).all()
print(results)