diff options
author | Mike Bayer <mike_mp@zzzcomputing.com> | 2022-08-18 13:56:50 -0400 |
---|---|---|
committer | Mike Bayer <mike_mp@zzzcomputing.com> | 2022-09-25 19:38:10 -0400 |
commit | 81d8394c0b5342cdc603cb2e07e12139c9506bf6 (patch) | |
tree | 5453f51ef80bb3b0b4705025070439fdccfea29c /examples/inheritance/concrete.py | |
parent | a8029f5a7e3e376ec57f1614ab0294b717d53c05 (diff) | |
download | sqlalchemy-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.py | 234 |
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() |