Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1#!/usr/bin/env python 

2# cardinal_pythonlib/sqlalchemy/orm_inspect.py 

3 

4""" 

5=============================================================================== 

6 

7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com). 

8 

9 This file is part of cardinal_pythonlib. 

10 

11 Licensed under the Apache License, Version 2.0 (the "License"); 

12 you may not use this file except in compliance with the License. 

13 You may obtain a copy of the License at 

14 

15 https://www.apache.org/licenses/LICENSE-2.0 

16 

17 Unless required by applicable law or agreed to in writing, software 

18 distributed under the License is distributed on an "AS IS" BASIS, 

19 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

20 See the License for the specific language governing permissions and 

21 limitations under the License. 

22 

23=============================================================================== 

24 

25**Functions to inspect and copy SQLAlchemy ORM objects.** 

26 

27""" 

28 

29from typing import (Dict, Generator, List, Set, Tuple, Type, TYPE_CHECKING, 

30 Union) 

31 

32# noinspection PyProtectedMember 

33from sqlalchemy.ext.declarative.base import _get_immediate_cls_attr 

34from sqlalchemy.inspection import inspect 

35from sqlalchemy.orm.base import class_mapper 

36from sqlalchemy.orm.mapper import Mapper 

37from sqlalchemy.orm.relationships import RelationshipProperty 

38from sqlalchemy.orm.session import Session 

39from sqlalchemy.sql.schema import Column, MetaData 

40from sqlalchemy.sql.type_api import TypeEngine 

41from sqlalchemy.sql.visitors import VisitableType 

42from sqlalchemy.util import OrderedProperties 

43 

44from cardinal_pythonlib.classes import gen_all_subclasses 

45from cardinal_pythonlib.enumlike import OrderedNamespace 

46from cardinal_pythonlib.dicts import reversedict 

47from cardinal_pythonlib.logs import get_brace_style_log_with_null_handler 

48 

49if TYPE_CHECKING: 

50 from sqlalchemy.orm.state import InstanceState 

51 from sqlalchemy.sql.schema import Table 

52 

53log = get_brace_style_log_with_null_handler(__name__) 

54 

55 

56# ============================================================================= 

57# Creating ORM objects conveniently, etc. 

58# ============================================================================= 

59 

60def coltype_as_typeengine(coltype: Union[VisitableType, 

61 TypeEngine]) -> TypeEngine: 

62 """ 

63 Instances of SQLAlchemy column types are subclasses of ``TypeEngine``. 

64 It's possible to specify column types either as such instances, or as the 

65 class type. This function ensures that such classes are converted to 

66 instances. 

67 

68 To explain: you can specify columns like 

69 

70 .. code-block:: python 

71 

72 a = Column("a", Integer) 

73 b = Column("b", Integer()) 

74 c = Column("c", String(length=50)) 

75 

76 isinstance(Integer, TypeEngine) # False 

77 isinstance(Integer(), TypeEngine) # True 

78 isinstance(String(length=50), TypeEngine) # True 

79 

80 type(Integer) # <class 'sqlalchemy.sql.visitors.VisitableType'> 

81 type(Integer()) # <class 'sqlalchemy.sql.sqltypes.Integer'> 

82 type(String) # <class 'sqlalchemy.sql.visitors.VisitableType'> 

83 type(String(length=50)) # <class 'sqlalchemy.sql.sqltypes.String'> 

84 

85 This function coerces things to a :class:`TypeEngine`. 

86 """ 

87 if isinstance(coltype, TypeEngine): 

88 return coltype 

89 return coltype() # type: TypeEngine 

90 

91 

92# ============================================================================= 

93# SqlAlchemyAttrDictMixin 

94# ============================================================================= 

95 

96class SqlAlchemyAttrDictMixin(object): 

97 """ 

98 Mixin to: 

99 

100 - get a plain dictionary-like object (with attributes so we can use ``x.y`` 

101 rather than ``x['y']``) from an SQLAlchemy ORM object 

102 - make a nice ``repr()`` default, maintaining field order 

103 

104 See https://stackoverflow.com/questions/2537471 and in particular 

105 https://stackoverflow.com/questions/2441796. 

106 """ 

107 

108 def get_attrdict(self) -> OrderedNamespace: 

109 """ 

110 Returns what looks like a plain object with the values of the 

111 SQLAlchemy ORM object. 

112 """ 

113 # noinspection PyUnresolvedReferences 

114 columns = self.__table__.columns.keys() 

115 values = (getattr(self, x) for x in columns) 

116 zipped = zip(columns, values) 

117 return OrderedNamespace(zipped) 

118 

119 def __repr__(self) -> str: 

120 return "<{classname}({kvp})>".format( 

121 classname=type(self).__name__, 

122 kvp=", ".join(f"{k}={v!r}" 

123 for k, v in self.get_attrdict().items()) 

124 ) 

125 

126 @classmethod 

127 def from_attrdict(cls, attrdict: OrderedNamespace) -> object: 

128 """ 

129 Builds a new instance of the ORM object from values in an attrdict. 

130 """ 

131 dictionary = attrdict.__dict__ 

132 # noinspection PyArgumentList 

133 return cls(**dictionary) 

134 

135 

136# ============================================================================= 

137# Traverse ORM relationships (SQLAlchemy ORM) 

138# ============================================================================= 

139 

140def walk_orm_tree(obj, 

141 debug: bool = False, 

142 seen: Set = None, 

143 skip_relationships_always: List[str] = None, 

144 skip_relationships_by_tablename: Dict[str, List[str]] = None, 

145 skip_all_relationships_for_tablenames: List[str] = None, 

146 skip_all_objects_for_tablenames: List[str] = None) \ 

147 -> Generator[object, None, None]: 

148 """ 

149 Starting with a SQLAlchemy ORM object, this function walks a 

150 relationship tree, yielding each of the objects once. 

151 

152 To skip attributes by name, put the attribute name(s) in 

153 ``skip_attrs_always``. To skip by table name, pass 

154 ``skip_attrs_by_tablename`` as e.g. 

155 

156 .. code-block:: python 

157 

158 {'sometable': ['attr1_to_skip', 'attr2_to_skip']} 

159 

160 

161 Args: 

162 obj: the SQLAlchemy ORM object to walk 

163 

164 debug: be verbose 

165 

166 seen: usually ``None``, but can be a set of objects marked as "already 

167 seen"; if an object is in this set, it is skipped 

168 

169 skip_relationships_always: relationships are skipped if the 

170 relationship has a name in this (optional) list 

171 

172 skip_relationships_by_tablename: optional dictionary mapping table 

173 names (keys) to relationship attribute names (values); if the 

174 "related table"/"relationship attribute" pair are in this 

175 dictionary, the relationship is skipped 

176 

177 skip_all_relationships_for_tablenames: relationships are skipped if the 

178 the related table has a name in this (optional) list 

179 

180 skip_all_objects_for_tablenames: if the object belongs to a table whose 

181 name is in this (optional) list, the object is skipped 

182 

183 Yields: 

184 SQLAlchemy ORM objects (including the starting object) 

185 

186 """ 

187 # http://docs.sqlalchemy.org/en/latest/faq/sessions.html#faq-walk-objects 

188 skip_relationships_always = skip_relationships_always or [] # type: List[str] # noqa 

189 skip_relationships_by_tablename = skip_relationships_by_tablename or {} # type: Dict[str, List[str]] # noqa 

190 skip_all_relationships_for_tablenames = skip_all_relationships_for_tablenames or [] # type: List[str] # noqa 

191 skip_all_objects_for_tablenames = skip_all_objects_for_tablenames or [] # type: List[str] # noqa 

192 stack = [obj] 

193 if seen is None: 

194 seen = set() 

195 while stack: 

196 obj = stack.pop(0) 

197 if obj in seen: 

198 continue 

199 tablename = obj.__tablename__ 

200 if tablename in skip_all_objects_for_tablenames: 

201 continue 

202 seen.add(obj) 

203 if debug: 

204 log.debug("walk: yielding {!r}", obj) 

205 yield obj 

206 insp = inspect(obj) # type: InstanceState 

207 for relationship in insp.mapper.relationships: # type: RelationshipProperty # noqa 

208 attrname = relationship.key 

209 # Skip? 

210 if attrname in skip_relationships_always: 

211 continue 

212 if tablename in skip_all_relationships_for_tablenames: 

213 continue 

214 if (tablename in skip_relationships_by_tablename and 

215 attrname in skip_relationships_by_tablename[tablename]): 

216 continue 

217 # Process relationship 

218 if debug: 

219 log.debug("walk: following relationship {}", relationship) 

220 related = getattr(obj, attrname) 

221 if debug and related: 

222 log.debug("walk: queueing {!r}", related) 

223 if relationship.uselist: 

224 stack.extend(related) 

225 elif related is not None: 

226 stack.append(related) 

227 

228 

229# ============================================================================= 

230# deepcopy an SQLAlchemy object 

231# ============================================================================= 

232# Use case: object X is in the database; we want to clone it to object Y, 

233# which we can then save to the database, i.e. copying all SQLAlchemy field 

234# attributes of X except its PK. We also want it to copy anything that is 

235# dependent upon X, i.e. traverse relationships. 

236# 

237# https://groups.google.com/forum/#!topic/sqlalchemy/wb2M_oYkQdY 

238# https://groups.google.com/forum/#!searchin/sqlalchemy/cascade%7Csort:date/sqlalchemy/eIOkkXwJ-Ms/JLnpI2wJAAAJ # noqa 

239 

240def copy_sqla_object(obj: object, 

241 omit_fk: bool = True, 

242 omit_pk: bool = True, 

243 omit_attrs: List[str] = None, 

244 debug: bool = False) -> object: 

245 """ 

246 Given an SQLAlchemy object, creates a new object (FOR WHICH THE OBJECT 

247 MUST SUPPORT CREATION USING ``__init__()`` WITH NO PARAMETERS), and copies 

248 across all attributes, omitting PKs (by default), FKs (by default), and 

249 relationship attributes (always omitted). 

250 

251 Args: 

252 obj: the object to copy 

253 omit_fk: omit foreign keys (FKs)? 

254 omit_pk: omit primary keys (PKs)? 

255 omit_attrs: attributes (by name) not to copy 

256 debug: be verbose 

257  

258 Returns: 

259 a new copy of the object 

260 """ 

261 omit_attrs = omit_attrs or [] # type: List[str] 

262 cls = type(obj) 

263 mapper = class_mapper(cls) 

264 newobj = cls() # not: cls.__new__(cls) 

265 rel_keys = set([c.key for c in mapper.relationships]) 

266 prohibited = rel_keys 

267 if omit_pk: 

268 pk_keys = set([c.key for c in mapper.primary_key]) 

269 prohibited |= pk_keys 

270 if omit_fk: 

271 fk_keys = set([c.key for c in mapper.columns if c.foreign_keys]) 

272 prohibited |= fk_keys 

273 prohibited |= set(omit_attrs) 

274 if debug: 

275 log.debug("copy_sqla_object: skipping: {}", prohibited) 

276 for k in [p.key for p in mapper.iterate_properties 

277 if p.key not in prohibited]: 

278 try: 

279 value = getattr(obj, k) 

280 if debug: 

281 log.debug("copy_sqla_object: processing attribute {} = {}", 

282 k, value) 

283 setattr(newobj, k, value) 

284 except AttributeError: 

285 if debug: 

286 log.debug("copy_sqla_object: failed attribute {}", k) 

287 pass 

288 return newobj 

289 

290 

291def rewrite_relationships(oldobj: object, 

292 newobj: object, 

293 objmap: Dict[object, object], 

294 debug: bool = False, 

295 skip_table_names: List[str] = None) -> None: 

296 """ 

297 A utility function only. 

298 Used in copying objects between SQLAlchemy sessions. 

299 

300 Both ``oldobj`` and ``newobj`` are SQLAlchemy instances. The instance 

301 ``newobj`` is already a copy of ``oldobj`` but we wish to rewrite its 

302 relationships, according to the map ``objmap``, which maps old to new 

303 objects. 

304 

305 For example: 

306 

307 - Suppose a source session has a Customer record and a Sale record 

308 containing ``sale.customer_id``, a foreign key to Customer. 

309 

310 - We may have corresponding Python SQLAlchemy ORM objects 

311 ``customer_1_src`` and ``sale_1_src``. 

312 

313 - We copy them into a destination database, where their Python ORM objects 

314 are ``customer_1_dest`` and ``sale_1_dest``. 

315 

316 - In the process we set up an object map looking like: 

317 

318 .. code-block:: none 

319 

320 Old session New session 

321 ------------------------------- 

322 customer_1_src customer_1_dest 

323 sale_1_src sale_1_dest 

324 

325 - Now, we wish to make ``sale_1_dest`` have a relationship to 

326 ``customer_1_dest``, in the same way that ``sale_1_src`` has a 

327 relationship to ``customer_1_src``. This function will modify 

328 ``sale_1_dest`` accordingly, given this object map. It will observe that 

329 ``sale_1_src`` (here ``oldobj``) has a relationship to 

330 ``customer_1_src``; it will note that ``objmap`` maps ``customer_1_src`` 

331 to ``customer_1_dest``; it will create the relationship from 

332 ``sale_1_dest`` (here ``newobj``) to ``customer_1_dest``. 

333 

334 Args: 

335 oldobj: SQLAlchemy ORM object to read from 

336 

337 newobj: SQLAlchemy ORM object to write to 

338 

339 objmap: dictionary mapping "source" objects to their corresponding 

340 "destination" object. 

341 

342 debug: be verbose 

343 

344 skip_table_names: if a related table's name is in this (optional) list, 

345 that relationship is skipped 

346 """ 

347 skip_table_names = skip_table_names or [] # type: List[str] 

348 insp = inspect(oldobj) # type: InstanceState 

349 # insp.mapper.relationships is of type 

350 # sqlalchemy.utils._collections.ImmutableProperties, which is basically 

351 # a sort of AttrDict. 

352 for attrname_rel in insp.mapper.relationships.items(): # type: Tuple[str, RelationshipProperty] # noqa 

353 attrname = attrname_rel[0] 

354 rel_prop = attrname_rel[1] 

355 if rel_prop.viewonly: 

356 if debug: 

357 log.debug("Skipping viewonly relationship") 

358 continue # don't attempt to write viewonly relationships # noqa 

359 related_class = rel_prop.mapper.class_ 

360 related_table_name = related_class.__tablename__ # type: str 

361 if related_table_name in skip_table_names: 

362 if debug: 

363 log.debug("Skipping relationship for related table {!r}", 

364 related_table_name) 

365 continue 

366 # The relationship is an abstract object (so getting the 

367 # relationship from the old object and from the new, with e.g. 

368 # newrel = newinsp.mapper.relationships[oldrel.key], 

369 # yield the same object. All we need from it is the key name. 

370 # rel_key = rel.key # type: str 

371 # ... but also available from the mapper as attrname, above 

372 related_old = getattr(oldobj, attrname) 

373 if rel_prop.uselist: 

374 related_new = [objmap[r] for r in related_old] 

375 elif related_old is not None: 

376 related_new = objmap[related_old] 

377 else: 

378 related_new = None 

379 if debug: 

380 log.debug("rewrite_relationships: relationship {} -> {}", 

381 attrname, related_new) 

382 setattr(newobj, attrname, related_new) 

383 

384 

385def deepcopy_sqla_objects( 

386 startobjs: List[object], 

387 session: Session, 

388 flush: bool = True, 

389 debug: bool = False, 

390 debug_walk: bool = True, 

391 debug_rewrite_rel: bool = False, 

392 objmap: Dict[object, object] = None) -> None: 

393 """ 

394 Makes a copy of the specified SQLAlchemy ORM objects, inserting them into a 

395 new session. 

396 

397 This function operates in several passes: 

398 

399 1. Walk the ORM tree through all objects and their relationships, copying 

400 every object thus found (via :func:`copy_sqla_object`, without their 

401 relationships), and building a map from each source-session object to 

402 its equivalent destination-session object. 

403 

404 2. Work through all the destination objects, rewriting their relationships 

405 (via :func:`rewrite_relationships`) so they relate to each other (rather 

406 than their source-session brethren). 

407 

408 3. Insert all the destination-session objects into the destination session. 

409 

410 For this to succeed, every object must take an ``__init__`` call with no 

411 arguments (see :func:`copy_sqla_object`). (We can't specify the required 

412 ``args``/``kwargs``, since we are copying a tree of arbitrary objects.) 

413 

414 Args: 

415 startobjs: SQLAlchemy ORM objects to copy 

416 session: destination SQLAlchemy :class:`Session` into which to insert 

417 the copies 

418 flush: flush the session when we've finished? 

419 debug: be verbose? 

420 debug_walk: be extra verbose when walking the ORM tree? 

421 debug_rewrite_rel: be extra verbose when rewriting relationships? 

422 objmap: starting object map from source-session to destination-session 

423 objects (see :func:`rewrite_relationships` for more detail); 

424 usually ``None`` to begin with. 

425 """ 

426 if objmap is None: 

427 objmap = {} # keys = old objects, values = new objects 

428 if debug: 

429 log.debug("deepcopy_sqla_objects: pass 1: create new objects") 

430 

431 # Pass 1: iterate through all objects. (Can't guarantee to get 

432 # relationships correct until we've done this, since we don't know whether 

433 # or where the "root" of the PK tree is.) 

434 seen = set() 

435 for startobj in startobjs: 

436 for oldobj in walk_orm_tree(startobj, seen=seen, debug=debug_walk): 

437 if debug: 

438 log.debug("deepcopy_sqla_objects: copying {}", oldobj) 

439 newobj = copy_sqla_object(oldobj, omit_pk=True, omit_fk=True) 

440 # Don't insert the new object into the session here; it may trigger 

441 # an autoflush as the relationships are queried, and the new 

442 # objects are not ready for insertion yet (as their relationships 

443 # aren't set). 

444 # Note also the session.no_autoflush option: 

445 # "sqlalchemy.exc.OperationalError: (raised as a result of Query- 

446 # invoked autoflush; consider using a session.no_autoflush block if 

447 # this flush is occurring prematurely)..." 

448 objmap[oldobj] = newobj 

449 

450 # Pass 2: set all relationship properties. 

451 if debug: 

452 log.debug("deepcopy_sqla_objects: pass 2: set relationships") 

453 for oldobj, newobj in objmap.items(): 

454 if debug: 

455 log.debug("deepcopy_sqla_objects: newobj: {}", newobj) 

456 rewrite_relationships(oldobj, newobj, objmap, debug=debug_rewrite_rel) 

457 

458 # Now we can do session insert. 

459 if debug: 

460 log.debug("deepcopy_sqla_objects: pass 3: insert into session") 

461 for newobj in objmap.values(): 

462 session.add(newobj) 

463 

464 # Done 

465 if debug: 

466 log.debug("deepcopy_sqla_objects: done") 

467 if flush: 

468 session.flush() 

469 

470 

471def deepcopy_sqla_object(startobj: object, 

472 session: Session, 

473 flush: bool = True, 

474 debug: bool = False, 

475 debug_walk: bool = False, 

476 debug_rewrite_rel: bool = False, 

477 objmap: Dict[object, object] = None) -> object: 

478 """ 

479 Makes a copy of the object, inserting it into ``session``. 

480 

481 Uses :func:`deepcopy_sqla_objects` (q.v.). 

482 

483 A problem is the creation of duplicate dependency objects if you call it 

484 repeatedly. 

485 

486 Optionally, if you pass the objmap in (which maps old to new objects), you 

487 can call this function repeatedly to clone a related set of objects... 

488 ... no, that doesn't really work, as it doesn't visit parents before 

489 children. The :func:`cardinal_pythonlib.sqlalchemy.merge_db.merge_db` 

490 function does that properly. 

491 

492 Args: 

493 startobj: SQLAlchemy ORM object to deep-copy 

494 session: see :func:`deepcopy_sqla_objects` 

495 flush: see :func:`deepcopy_sqla_objects` 

496 debug: see :func:`deepcopy_sqla_objects` 

497 debug_walk: see :func:`deepcopy_sqla_objects` 

498 debug_rewrite_rel: see :func:`deepcopy_sqla_objects` 

499 objmap: see :func:`deepcopy_sqla_objects` 

500 

501 Returns: 

502 the copied object matching ``startobj`` 

503 

504 """ 

505 if objmap is None: 

506 objmap = {} # keys = old objects, values = new objects 

507 deepcopy_sqla_objects( 

508 startobjs=[startobj], 

509 session=session, 

510 flush=flush, 

511 debug=debug, 

512 debug_walk=debug_walk, 

513 debug_rewrite_rel=debug_rewrite_rel, 

514 objmap=objmap 

515 ) 

516 return objmap[startobj] # returns the new object matching startobj 

517 

518 

519# ============================================================================= 

520# Get Columns from an ORM instance 

521# ============================================================================= 

522 

523def gen_columns(obj) -> Generator[Tuple[str, Column], None, None]: 

524 """ 

525 Asks a SQLAlchemy ORM object: "what are your SQLAlchemy columns?" 

526 

527 Yields tuples of ``(attr_name, Column)`` from an SQLAlchemy ORM object 

528 instance. Also works with the corresponding SQLAlchemy ORM class. Examples: 

529 

530 .. code-block:: python 

531 

532 from sqlalchemy.ext.declarative import declarative_base 

533 from sqlalchemy.sql.schema import Column 

534 from sqlalchemy.sql.sqltypes import Integer 

535 

536 Base = declarative_base() 

537 

538 class MyClass(Base): 

539 __tablename__ = "mytable" 

540 pk = Column("pk", Integer, primary_key=True, autoincrement=True) 

541 a = Column("a", Integer) 

542 

543 x = MyClass() 

544 

545 list(gen_columns(x)) 

546 list(gen_columns(MyClass)) 

547 

548 """ 

549 mapper = obj.__mapper__ # type: Mapper 

550 assert mapper, ( 

551 f"gen_columns called on {obj!r} which is not an SQLAlchemy ORM object" 

552 ) 

553 colmap = mapper.columns # type: OrderedProperties 

554 if not colmap: 

555 return 

556 for attrname, column in colmap.items(): 

557 # NB: column.name is the SQL column name, not the attribute name 

558 yield attrname, column 

559 # Don't bother using 

560 # cls = obj.__class_ 

561 # for attrname in dir(cls): 

562 # cls_attr = getattr(cls, attrname) 

563 # # ... because, for columns, these will all be instances of 

564 # # sqlalchemy.orm.attributes.InstrumentedAttribute. 

565 

566 

567def get_pk_attrnames(obj) -> List[str]: 

568 """ 

569 Asks an SQLAlchemy ORM object: "what are your primary key(s)?" 

570 

571 Args: 

572 obj: SQLAlchemy ORM object 

573 

574 Returns: 

575 list of attribute names of primary-key columns 

576 

577 """ 

578 return [attrname 

579 for attrname, column in gen_columns(obj) 

580 if column.primary_key] 

581 

582 

583def gen_columns_for_uninstrumented_class(cls: Type) \ 

584 -> Generator[Tuple[str, Column], None, None]: 

585 """ 

586 Generate ``(attr_name, Column)`` tuples from an UNINSTRUMENTED class, i.e. 

587 one that does not inherit from ``declarative_base()``. Use this for mixins 

588 of that kind. 

589 

590 SUBOPTIMAL. May produce warnings like: 

591  

592 .. code-block:: none 

593  

594 SAWarning: Unmanaged access of declarative attribute id from non-mapped class GenericTabletRecordMixin 

595 

596 Try to use :func:`gen_columns` instead. 

597 """ # noqa 

598 for attrname in dir(cls): 

599 potential_column = getattr(cls, attrname) 

600 if isinstance(potential_column, Column): 

601 yield attrname, potential_column 

602 

603 

604def attrname_to_colname_dict(cls) -> Dict[str, str]: 

605 """ 

606 Asks an SQLAlchemy class how its attribute names correspond to database 

607 column names. 

608 

609 Args: 

610 cls: SQLAlchemy ORM class 

611 

612 Returns: 

613 a dictionary mapping attribute names to database column names 

614 """ 

615 attr_col = {} # type: Dict[str, str] 

616 for attrname, column in gen_columns(cls): 

617 attr_col[attrname] = column.name 

618 return attr_col 

619 

620 

621def colname_to_attrname_dict(cls) -> Dict[str, str]: 

622 return reversedict(attrname_to_colname_dict(cls)) 

623 

624 

625# ============================================================================= 

626# Get relationships from an ORM instance 

627# ============================================================================= 

628 

629def gen_relationships(obj) -> Generator[Tuple[str, RelationshipProperty, Type], 

630 None, None]: 

631 """ 

632 Yields tuples of ``(attrname, RelationshipProperty, related_class)`` 

633 for all relationships of an ORM object. 

634 The object 'obj' can be EITHER an instance OR a class. 

635 """ 

636 insp = inspect(obj) # type: InstanceState 

637 # insp.mapper.relationships is of type 

638 # sqlalchemy.utils._collections.ImmutableProperties, which is basically 

639 # a sort of AttrDict. 

640 for attrname, rel_prop in insp.mapper.relationships.items(): # type: Tuple[str, RelationshipProperty] # noqa 

641 # noinspection PyUnresolvedReferences 

642 related_class = rel_prop.mapper.class_ 

643 # log.critical("gen_relationships: attrname={!r}, " 

644 # "rel_prop={!r}, related_class={!r}, rel_prop.info={!r}", 

645 # attrname, rel_prop, related_class, rel_prop.info) 

646 yield attrname, rel_prop, related_class 

647 

648 

649# ============================================================================= 

650# Inspect ORM objects (SQLAlchemy ORM) 

651# ============================================================================= 

652 

653def get_orm_columns(cls: Type) -> List[Column]: 

654 """ 

655 Gets :class:`Column` objects from an SQLAlchemy ORM class. 

656 Does not provide their attribute names. 

657 """ 

658 mapper = inspect(cls) # type: Mapper 

659 # ... returns InstanceState if called with an ORM object 

660 # http://docs.sqlalchemy.org/en/latest/orm/session_state_management.html#session-object-states # noqa 

661 # ... returns Mapper if called with an ORM class 

662 # http://docs.sqlalchemy.org/en/latest/orm/mapping_api.html#sqlalchemy.orm.mapper.Mapper # noqa 

663 colmap = mapper.columns # type: OrderedProperties 

664 return colmap.values() 

665 

666 

667def get_orm_column_names(cls: Type, sort: bool = False) -> List[str]: 

668 """ 

669 Gets column names (that is, database column names) from an SQLAlchemy 

670 ORM class. 

671 """ 

672 colnames = [col.name for col in get_orm_columns(cls)] 

673 return sorted(colnames) if sort else colnames 

674 

675 

676# ============================================================================= 

677# Inspect metadata (SQLAlchemy ORM) 

678# ============================================================================= 

679 

680def get_table_names_from_metadata(metadata: MetaData) -> List[str]: 

681 """ 

682 Returns all database table names found in an SQLAlchemy :class:`MetaData` 

683 object. 

684 """ 

685 return [table.name for table in metadata.tables.values()] 

686 

687 

688def get_metadata_from_orm_class_or_object(cls: Type) -> MetaData: 

689 """ 

690 Returns the :class:`MetaData` object from an SQLAlchemy ORM class or 

691 instance. 

692 """ 

693 # noinspection PyUnresolvedReferences 

694 table = cls.__table__ # type: Table 

695 return table.metadata 

696 

697 

698def gen_orm_classes_from_base(base: Type) -> Generator[Type, None, None]: 

699 """ 

700 From an SQLAlchemy ORM base class, yield all the subclasses (except those 

701 that are abstract). 

702 

703 If you begin with the proper :class`Base` class, then this should give all 

704 ORM classes in use. 

705 """ 

706 for cls in gen_all_subclasses(base): 

707 if _get_immediate_cls_attr(cls, '__abstract__', strict=True): 

708 # This is SQLAlchemy's own way of detecting abstract classes; see 

709 # sqlalchemy.ext.declarative.base 

710 continue # NOT an ORM class 

711 yield cls 

712 

713 

714def get_orm_classes_by_table_name_from_base(base: Type) -> Dict[str, Type]: 

715 """ 

716 Given an SQLAlchemy ORM base class, returns a dictionary whose keys are 

717 table names and whose values are ORM classes. 

718 

719 If you begin with the proper :class`Base` class, then this should give all 

720 tables and ORM classes in use. 

721 """ 

722 # noinspection PyUnresolvedReferences 

723 return {cls.__tablename__: cls for cls in gen_orm_classes_from_base(base)}