(self, connection, schema=None, **kw)
| 5591 | |
| 5592 | @reflection.cache |
| 5593 | def _load_domains(self, connection, schema=None, **kw): |
| 5594 | result = connection.execute(self._domain_query(schema)) |
| 5595 | |
| 5596 | domains: List[ReflectedDomain] = [] |
| 5597 | for domain in result.mappings(): |
| 5598 | # strip (30) from character varying(30) |
| 5599 | attype = re.search(r"([^\(]+)", domain["attype"]).group(1) |
| 5600 | constraints: List[ReflectedDomainConstraint] = [] |
| 5601 | if domain["connames"]: |
| 5602 | # When a domain has multiple CHECK constraints, they will |
| 5603 | # be tested in alphabetical order by name. |
| 5604 | sorted_constraints = sorted( |
| 5605 | zip(domain["connames"], domain["condefs"]), |
| 5606 | key=lambda t: t[0], |
| 5607 | ) |
| 5608 | for name, def_ in sorted_constraints: |
| 5609 | # constraint is in the form "CHECK (expression)" |
| 5610 | # or "NOT NULL". Ignore the "NOT NULL" and |
| 5611 | # remove "CHECK (" and the tailing ")". |
| 5612 | if def_.casefold().startswith("check"): |
| 5613 | check = def_[7:-1] |
| 5614 | constraints.append({"name": name, "check": check}) |
| 5615 | domain_rec: ReflectedDomain = { |
| 5616 | "name": domain["name"], |
| 5617 | "schema": domain["schema"], |
| 5618 | "visible": domain["visible"], |
| 5619 | "type": attype, |
| 5620 | "nullable": domain["nullable"], |
| 5621 | "default": domain["default"], |
| 5622 | "constraints": constraints, |
| 5623 | "collation": domain["collname"], |
| 5624 | } |
| 5625 | domains.append(domain_rec) |
| 5626 | |
| 5627 | return domains |
| 5628 | |
| 5629 | @util.memoized_property |
| 5630 | def _pg_am_query(self): |
no test coverage detected