()
| 2657 | table_data = self._get_table_sql(connection, table_name, schema=schema) |
| 2658 | |
| 2659 | def parse_fks(): |
| 2660 | if table_data is None: |
| 2661 | # system tables, etc. |
| 2662 | return |
| 2663 | |
| 2664 | # note that we already have the FKs from PRAGMA above. This whole |
| 2665 | # regexp thing is trying to locate additional detail about the |
| 2666 | # FKs, namely the name of the constraint and other options. |
| 2667 | # so parsing the columns is really about matching it up to what |
| 2668 | # we already have. |
| 2669 | FK_PATTERN = ( |
| 2670 | r'(?:CONSTRAINT +(?:"(.+?)"|(\w+)) +)?' |
| 2671 | r"FOREIGN KEY *\( *(.+?) *\) +" |
| 2672 | r'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\( *((?:(?:"[^"]+"|[a-z0-9_]+) *(?:, *)?)+)\) *' # noqa: E501 |
| 2673 | r"((?:ON (?:DELETE|UPDATE) " |
| 2674 | r"(?:SET NULL|SET DEFAULT|CASCADE|RESTRICT|NO ACTION) *)*)" |
| 2675 | r"((?:NOT +)?DEFERRABLE)?" |
| 2676 | r"(?: +INITIALLY +(DEFERRED|IMMEDIATE))?" |
| 2677 | ) |
| 2678 | for match in re.finditer(FK_PATTERN, table_data, re.I): |
| 2679 | ( |
| 2680 | constraint_quoted_name, |
| 2681 | constraint_name, |
| 2682 | constrained_columns, |
| 2683 | referred_quoted_name, |
| 2684 | referred_name, |
| 2685 | referred_columns, |
| 2686 | onupdatedelete, |
| 2687 | deferrable, |
| 2688 | initially, |
| 2689 | ) = match.group(1, 2, 3, 4, 5, 6, 7, 8, 9) |
| 2690 | constraint_name = constraint_quoted_name or constraint_name |
| 2691 | constrained_columns = list( |
| 2692 | self._find_cols_in_sig(constrained_columns) |
| 2693 | ) |
| 2694 | if not referred_columns: |
| 2695 | referred_columns = constrained_columns |
| 2696 | else: |
| 2697 | referred_columns = list( |
| 2698 | self._find_cols_in_sig(referred_columns) |
| 2699 | ) |
| 2700 | referred_name = referred_quoted_name or referred_name |
| 2701 | options = {} |
| 2702 | |
| 2703 | for token in re.split(r" *\bON\b *", onupdatedelete.upper()): |
| 2704 | if token.startswith("DELETE"): |
| 2705 | ondelete = token[6:].strip() |
| 2706 | if ondelete and ondelete != "NO ACTION": |
| 2707 | options["ondelete"] = ondelete |
| 2708 | elif token.startswith("UPDATE"): |
| 2709 | onupdate = token[6:].strip() |
| 2710 | if onupdate and onupdate != "NO ACTION": |
| 2711 | options["onupdate"] = onupdate |
| 2712 | |
| 2713 | if deferrable: |
| 2714 | options["deferrable"] = "NOT" not in deferrable.upper() |
| 2715 | if initially: |
| 2716 | options["initially"] = initially.upper() |
nothing calls this directly
no test coverage detected