(self)
| 1959 | ) |
| 1960 | |
| 1961 | def test_basic(self): |
| 1962 | items = self.tables.items |
| 1963 | |
| 1964 | class Container(BasicEntity): |
| 1965 | pass |
| 1966 | |
| 1967 | class LineItem(BasicEntity): |
| 1968 | pass |
| 1969 | |
| 1970 | container_select = ( |
| 1971 | sa.select(items.c.policyNum, items.c.policyEffDate, items.c.type) |
| 1972 | .distinct() |
| 1973 | .alias("container_select") |
| 1974 | ) |
| 1975 | |
| 1976 | self.mapper_registry.map_imperatively(LineItem, items) |
| 1977 | |
| 1978 | self.mapper_registry.map_imperatively( |
| 1979 | Container, |
| 1980 | container_select, |
| 1981 | properties=dict( |
| 1982 | lineItems=relationship( |
| 1983 | LineItem, |
| 1984 | lazy="select", |
| 1985 | cascade="all, delete-orphan", |
| 1986 | order_by=sa.asc(items.c.id), |
| 1987 | primaryjoin=sa.and_( |
| 1988 | container_select.c.policyNum == items.c.policyNum, |
| 1989 | container_select.c.policyEffDate |
| 1990 | == items.c.policyEffDate, |
| 1991 | container_select.c.type == items.c.type, |
| 1992 | ), |
| 1993 | foreign_keys=[ |
| 1994 | items.c.policyNum, |
| 1995 | items.c.policyEffDate, |
| 1996 | items.c.type, |
| 1997 | ], |
| 1998 | ) |
| 1999 | ), |
| 2000 | ) |
| 2001 | |
| 2002 | session = fixture_session() |
| 2003 | con = Container() |
| 2004 | con.policyNum = "99" |
| 2005 | con.policyEffDate = datetime.date.today() |
| 2006 | con.type = "TESTER" |
| 2007 | session.add(con) |
| 2008 | for i in range(0, 10): |
| 2009 | li = LineItem() |
| 2010 | li.id = i |
| 2011 | con.lineItems.append(li) |
| 2012 | session.add(li) |
| 2013 | session.flush() |
| 2014 | session.expunge_all() |
| 2015 | newcon = ( |
| 2016 | session.query(Container).order_by(container_select.c.type).first() |
| 2017 | ) |
| 2018 | assert con.policyNum == newcon.policyNum |
nothing calls this directly
no test coverage detected