summaryrefslogtreecommitdiff
path: root/examples/inheritance/concrete.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2022-08-18 13:56:50 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2022-09-25 19:38:10 -0400
commit81d8394c0b5342cdc603cb2e07e12139c9506bf6 (patch)
tree5453f51ef80bb3b0b4705025070439fdccfea29c /examples/inheritance/concrete.py
parenta8029f5a7e3e376ec57f1614ab0294b717d53c05 (diff)
downloadsqlalchemy-81d8394c0b5342cdc603cb2e07e12139c9506bf6.tar.gz
New ORM Query Guide featuring DML support
reviewers: these docs publish periodically at: https://docs.sqlalchemy.org/en/gerrit/4042/orm/queryguide/index.html See the "last generated" timestamp near the bottom of the page to ensure the latest version is up Change includes some other adjustments: * small typing fixes for end-user benefit * removal of a bunch of old examples for patterns that nobody uses or aren't really what we promote now * modernization of some examples, including inheritance Change-Id: I9929daab7797be9515f71c888b28af1209e789ff
Diffstat (limited to 'examples/inheritance/concrete.py')
-rw-r--r--examples/inheritance/concrete.py234
1 files changed, 116 insertions, 118 deletions
diff --git a/examples/inheritance/concrete.py b/examples/inheritance/concrete.py
index 4eb89984a..899b045b1 100644
--- a/examples/inheritance/concrete.py
+++ b/examples/inheritance/concrete.py
@@ -1,171 +1,169 @@
"""Concrete-table (table-per-class) inheritance example."""
+from __future__ import annotations
+
+from typing import Annotated
-from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import ForeignKey
-from sqlalchemy import inspect
-from sqlalchemy import Integer
from sqlalchemy import or_
+from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.ext.declarative import ConcreteBase
-from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import with_polymorphic
-Base = declarative_base()
+intpk = Annotated[int, mapped_column(primary_key=True)]
+str50 = Annotated[str, mapped_column(String(50))]
+
+
+class Base(DeclarativeBase):
+ pass
class Company(Base):
__tablename__ = "company"
- id = Column(Integer, primary_key=True)
- name = Column(String(50))
+ id: Mapped[intpk]
+ name: Mapped[str50]
- employees = relationship(
- "Person", back_populates="company", cascade="all, delete-orphan"
+ employees: Mapped[list[Person]] = relationship(
+ back_populates="company", cascade="all, delete-orphan"
)
def __repr__(self):
- return "Company %s" % self.name
+ return f"Company {self.name}"
class Person(ConcreteBase, Base):
__tablename__ = "person"
- id = Column(Integer, primary_key=True)
- company_id = Column(ForeignKey("company.id"))
- name = Column(String(50))
+ id: Mapped[intpk]
+ company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
+ name: Mapped[str50]
- company = relationship("Company", back_populates="employees")
+ company: Mapped[Company] = relationship(back_populates="employees")
- __mapper_args__ = {"polymorphic_identity": "person"}
+ __mapper_args__ = {
+ "polymorphic_identity": "person",
+ }
def __repr__(self):
- return "Ordinary person %s" % self.name
+ return f"Ordinary person {self.name}"
class Engineer(Person):
__tablename__ = "engineer"
- id = Column(Integer, primary_key=True)
- name = Column(String(50))
- company_id = Column(ForeignKey("company.id"))
- status = Column(String(30))
- engineer_name = Column(String(30))
- primary_language = Column(String(30))
- company = relationship("Company", back_populates="employees")
+ id: Mapped[int] = mapped_column(primary_key=True)
+ company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
+ name: Mapped[str50]
+ status: Mapped[str50]
+ engineer_name: Mapped[str50]
+ primary_language: Mapped[str50]
- __mapper_args__ = {"polymorphic_identity": "engineer", "concrete": True}
+ company: Mapped[Company] = relationship(back_populates="employees")
- def __repr__(self):
- return (
- "Engineer %s, status %s, engineer_name %s, "
- "primary_language %s"
- % (
- self.name,
- self.status,
- self.engineer_name,
- self.primary_language,
- )
- )
+ __mapper_args__ = {"polymorphic_identity": "engineer", "concrete": True}
class Manager(Person):
__tablename__ = "manager"
- id = Column(Integer, primary_key=True)
- name = Column(String(50))
- company_id = Column(ForeignKey("company.id"))
- status = Column(String(30))
- manager_name = Column(String(30))
- company = relationship("Company", back_populates="employees")
+ id: Mapped[int] = mapped_column(primary_key=True)
+ company_id: Mapped[int] = mapped_column(ForeignKey("company.id"))
+ name: Mapped[str50]
+ status: Mapped[str50]
+ manager_name: Mapped[str50]
+
+ company: Mapped[Company] = relationship(back_populates="employees")
__mapper_args__ = {"polymorphic_identity": "manager", "concrete": True}
def __repr__(self):
- return "Manager %s, status %s, manager_name %s" % (
- self.name,
- self.status,
- self.manager_name,
+ return (
+ f"Manager {self.name}, status {self.status}, "
+ f"manager_name {self.manager_name}"
)
engine = create_engine("sqlite://", echo=True)
Base.metadata.create_all(engine)
-session = Session(engine)
-
-c = Company(
- name="company1",
- employees=[
- Manager(
- name="pointy haired boss", status="AAB", manager_name="manager1"
- ),
- Engineer(
- name="dilbert",
- status="BBA",
- engineer_name="engineer1",
- primary_language="java",
- ),
- Person(name="joesmith"),
- Engineer(
- name="wally",
- status="CGG",
- engineer_name="engineer2",
- primary_language="python",
- ),
- Manager(name="jsmith", status="ABA", manager_name="manager2"),
- ],
-)
-session.add(c)
-
-session.commit()
-
-c = session.query(Company).get(1)
-for e in c.employees:
- print(e, inspect(e).key, e.company)
-assert set([e.name for e in c.employees]) == set(
- ["pointy haired boss", "dilbert", "joesmith", "wally", "jsmith"]
-)
-print("\n")
-
-dilbert = session.query(Person).filter_by(name="dilbert").one()
-dilbert2 = session.query(Engineer).filter_by(name="dilbert").one()
-assert dilbert is dilbert2
-
-dilbert.engineer_name = "hes dilbert!"
-
-session.commit()
-
-c = session.query(Company).get(1)
-for e in c.employees:
- print(e)
-
-# query using with_polymorphic.
-eng_manager = with_polymorphic(Person, [Engineer, Manager])
-print(
- session.query(eng_manager)
- .filter(
- or_(
- eng_manager.Engineer.engineer_name == "engineer1",
- eng_manager.Manager.manager_name == "manager2",
- )
+with Session(engine) as session:
+
+ c = Company(
+ name="company1",
+ employees=[
+ Manager(
+ name="mr krabs",
+ status="AAB",
+ manager_name="manager1",
+ ),
+ Engineer(
+ name="spongebob",
+ status="BBA",
+ engineer_name="engineer1",
+ primary_language="java",
+ ),
+ Person(name="joesmith"),
+ Engineer(
+ name="patrick",
+ status="CGG",
+ engineer_name="engineer2",
+ primary_language="python",
+ ),
+ Manager(name="jsmith", status="ABA", manager_name="manager2"),
+ ],
)
- .all()
-)
-
-# illustrate join from Company
-eng_manager = with_polymorphic(Person, [Engineer, Manager])
-print(
- session.query(Company)
- .join(Company.employees.of_type(eng_manager))
- .filter(
- or_(
- eng_manager.Engineer.engineer_name == "engineer1",
- eng_manager.Manager.manager_name == "manager2",
- )
+ session.add(c)
+
+ session.commit()
+
+ for e in c.employees:
+ print(e)
+
+ spongebob = session.scalars(
+ select(Person).filter_by(name="spongebob")
+ ).one()
+ spongebob2 = session.scalars(
+ select(Engineer).filter_by(name="spongebob")
+ ).one()
+ assert spongebob is spongebob2
+
+ spongebob2.engineer_name = "hes spongebob!"
+
+ session.commit()
+
+ # query using with_polymorphic.
+ # when using ConcreteBase, use "*" to use the default selectable
+ # setting specific entities won't work right now.
+ eng_manager = with_polymorphic(Person, "*")
+ print(
+ session.scalars(
+ select(eng_manager).filter(
+ or_(
+ eng_manager.Engineer.engineer_name == "engineer1",
+ eng_manager.Manager.manager_name == "manager2",
+ )
+ )
+ ).all()
+ )
+
+ # illustrate join from Company.
+ print(
+ session.scalars(
+ select(Company)
+ .join(Company.employees.of_type(eng_manager))
+ .filter(
+ or_(
+ eng_manager.Engineer.engineer_name == "engineer1",
+ eng_manager.Manager.manager_name == "manager2",
+ )
+ )
+ ).all()
)
- .all()
-)
-session.commit()
+ session.commit()