We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Refactor dependents tests to use library usecase
## Problem
Initially, both DB and API tests were created using dummy data and were quite difficult to read. 

## Proposed solution
Refactor them to use a common use case and simplify the cases.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `db/dependents/dependents_utils.py`
Content:
```
1 from sqlalchemy import MetaData, Table, any_, column, exists, func, literal, select, text, true, union
2 from sqlalchemy.dialects.postgresql import array
3 
4 # OIDs assigned during normal database operation are constrained to be 16384 or higher.
5 USER_DEFINED_OBJECTS_MIN_OID = 16384
6 # automatic and normal dependents
7 PG_DEPENDENT_TYPES = ['a', 'n']
8 PG_CLASS_CATALOGUE_NAME = '\'pg_class\''
9 START_LEVEL = 1
10 MAX_LEVEL = 10
11 
12 
13 def get_dependents_graph(referenced_object_id, engine):
14     dependency_pairs = _get_typed_dependency_pairs_stmt(engine)
15     dependency_pairs_cte = dependency_pairs.cte(recursive=True, name='dependency_pairs_cte')
16 
17     # anchor member which includes all dependents of a requested object
18     anchor = select(
19         dependency_pairs_cte,
20         literal(START_LEVEL).label('level'),
21         array([dependency_pairs_cte.c.refobjid]).label('dependency_chain')) \
22         .where(dependency_pairs_cte.c.refobjid == referenced_object_id) \
23         .where(dependency_pairs_cte.c.objid != referenced_object_id)
24     anchor = anchor.cte('cte')
25 
26     # recursive member which includes dependents for each object of the previous level
27     recursive = select(
28         dependency_pairs_cte,
29         (anchor.c.level + 1).label('level'),
30         anchor.c.dependency_chain + array([anchor.c.objid])) \
31         .where(anchor.c.level < MAX_LEVEL) \
32         .where(dependency_pairs_cte.c.objid != any_(anchor.c.dependency_chain)) \
33         .where(dependency_pairs_cte.c.objid != dependency_pairs_cte.c.refobjid)
34     recursive = recursive.join(anchor, dependency_pairs_cte.c.refobjid == anchor.c.objid)
35 
36     recursive_stmt = anchor.union(recursive)
37     stmt = select(recursive_stmt)
38 
39     with engine.connect() as conn:
40         result = conn.execute(stmt)
41 
42     return _get_structured_result(result)
43 
44 
45 # finding table dependents based on foreign key constraints from the referenced tables
46 def _get_table_dependents(foreign_key_dependents, pg_constraint_table):
47     # TODO: update refobjsubid with actual values when working on columns
48     pg_identify_object = _get_pg_identify_object_lateral_stmt(
49         text(f'{PG_CLASS_CATALOGUE_NAME}::regclass::oid'), pg_constraint_table.c.conrelid, 0)
50 
51     pg_identify_refobject = _get_pg_identify_object_lateral_stmt(
52         foreign_key_dependents.c.refclassid, foreign_key_dependents.c.refobjid, 0)
53 
54     # conrelid in this case is the oid of the table which a constraint resides in
55     return select(
56         foreign_key_dependents.c.classid,
57         pg_constraint_table.c.conrelid.label('objid'),
58         foreign_key_dependents.c.objsubid,
59         foreign_key_dependents.c.refclassid,
60         foreign_key_dependents.c.refobjid,
61         foreign_key_dependents.c.refobjsubid,
62         foreign_key_dependents.c.deptype,
63         pg_identify_object.c.name.label('objname'),
64         pg_identify_object.c.type.label('objtype'),
65         pg_identify_refobject.c.name.label('refobjname'),
66         pg_identify_refobject.c.type.label('refobjtype')) \
67         .select_from(foreign_key_dependents) \
68         .join(pg_constraint_table, pg_constraint_table.c.oid == foreign_key_dependents.c.objid) \
69         .join(pg_identify_object, true()) \
70         .join(pg_identify_refobject, true()) \
71         .where(pg_constraint_table.c.confrelid != 0) \
72         .group_by(
73             foreign_key_dependents,
74             pg_constraint_table.c.conrelid,
75             pg_identify_object.c.name,
76             pg_identify_object.c.type,
77             pg_identify_refobject.c.name,
78             pg_identify_refobject.c.type)
79 
80 
81 def _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair):
82     return dependency_pair.where(pg_identify_object.c.type == 'table constraint')
83 
84 
85 # getting a full list of dependents and identifying them
86 def _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject):
87     result = select(
88         pg_depend,
89         pg_identify_object.c.name.label('objname'),
90         pg_identify_object.c.type.label('objtype'),
91         pg_identify_refobject.c.name.label('refobjname'),
92         pg_identify_refobject.c.type.label('refobjtype')) \
93         .select_from(pg_depend) \
94         .join(pg_identify_object, true()) \
95         .join(pg_identify_refobject, true()) \
96         .where(pg_depend.c.deptype == any_(array(PG_DEPENDENT_TYPES))) \
97         .where(pg_depend.c.objid >= USER_DEFINED_OBJECTS_MIN_OID) \
98         .group_by(
99             pg_depend,
100             pg_identify_object.c.name,
101             pg_identify_object.c.type,
102             pg_identify_refobject.c.name,
103             pg_identify_refobject.c.type)
104 
105     return result
106 
107 
108 def _get_pg_depend_table(engine, metadata):
109     return Table("pg_depend", metadata, autoload_with=engine)
110 
111 
112 def _get_pg_constraint_table(engine, metadata):
113     return Table("pg_constraint", metadata, autoload_with=engine)
114 
115 
116 def _get_pg_identify_object_lateral_stmt(classid, objid, objsubid):
117     return select(
118         column("name"),
119         column("type")) \
120         .select_from(func.pg_identify_object(
121             classid,
122             objid,
123             objsubid)) \
124         .lateral()
125 
126 
127 def _get_typed_dependency_pairs_stmt(engine):
128     metadata = MetaData()
129 
130     pg_depend = _get_pg_depend_table(engine, metadata)
131     pg_identify_object = _get_pg_identify_object_lateral_stmt(
132         pg_depend.c.classid, pg_depend.c.objid, pg_depend.c.objsubid)
133     pg_identify_refobject = _get_pg_identify_object_lateral_stmt(
134         pg_depend.c.refclassid, pg_depend.c.refobjid, 0)
135     pg_constraint = _get_pg_constraint_table(engine, metadata)
136 
137     # each statement filters the base statement extracting dependents of a specific type
138     # so it's easy to exclude particular types or add new
139     dependency_pair = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)
140     foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair).cte('foreign_key_constraint_dependents')
141     table_dependents = _get_table_dependents(foreign_key_constraint_dependents, pg_constraint).cte('table_dependents')
142 
143     return union(
144         select(foreign_key_constraint_dependents),
145         select(table_dependents))
146 
147 
148 def has_dependencies(referenced_object_id, engine):
149     metadata = MetaData()
150 
151     pg_depend = _get_pg_depend_table(engine, metadata)
152 
153     stmt = select(
154         exists(
155             select().select_from(pg_depend)
156             .where(pg_depend.c.refobjid == referenced_object_id)
157             .where(pg_depend.c.deptype == any_(array(PG_DEPENDENT_TYPES)))
158             .where(pg_depend.c.objid >= USER_DEFINED_OBJECTS_MIN_OID)
159         )
160     )
161 
162     with engine.connect() as conn:
163         result = conn.execute(stmt).scalar()
164 
165     return result
166 
167 
168 def _get_structured_result(dependency_graph_result):
169     result = []
170     for dependency_pair in dependency_graph_result:
171         d = {}
172         d['level'] = dependency_pair.level
173         d['obj'] = {'objid': dependency_pair.objid, 'type': dependency_pair.objtype}
174         d['parent_obj'] = {'objid': dependency_pair.refobjid, 'type': dependency_pair.refobjtype}
175         result.append(d)
176 
177     return result
178
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/db/dependents/dependents_utils.py b/db/dependents/dependents_utils.py
--- a/db/dependents/dependents_utils.py
+++ b/db/dependents/dependents_utils.py
@@ -26,7 +26,7 @@
     # recursive member which includes dependents for each object of the previous level
     recursive = select(
         dependency_pairs_cte,
-        (anchor.c.level + 1).label('level'),
+        (anchor.c.level + 1),
         anchor.c.dependency_chain + array([anchor.c.objid])) \
         .where(anchor.c.level < MAX_LEVEL) \
         .where(dependency_pairs_cte.c.objid != any_(anchor.c.dependency_chain)) \
@@ -82,7 +82,7 @@
     return dependency_pair.where(pg_identify_object.c.type == 'table constraint')
 
 
-# getting a full list of dependents and identifying them
+# stmt for getting a full list of dependents and identifying them
 def _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject):
     result = select(
         pg_depend,
@@ -136,8 +136,8 @@
 
     # each statement filters the base statement extracting dependents of a specific type
     # so it's easy to exclude particular types or add new
-    dependency_pair = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)
-    foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair).cte('foreign_key_constraint_dependents')
+    dependency_pairs = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)
+    foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pairs).cte('foreign_key_constraint_dependents')
     table_dependents = _get_table_dependents(foreign_key_constraint_dependents, pg_constraint).cte('table_dependents')
 
     return union(

{"golden_diff": "diff --git a/db/dependents/dependents_utils.py b/db/dependents/dependents_utils.py\n--- a/db/dependents/dependents_utils.py\n+++ b/db/dependents/dependents_utils.py\n@@ -26,7 +26,7 @@\n     # recursive member which includes dependents for each object of the previous level\n     recursive = select(\n         dependency_pairs_cte,\n-        (anchor.c.level + 1).label('level'),\n+        (anchor.c.level + 1),\n         anchor.c.dependency_chain + array([anchor.c.objid])) \\\n         .where(anchor.c.level < MAX_LEVEL) \\\n         .where(dependency_pairs_cte.c.objid != any_(anchor.c.dependency_chain)) \\\n@@ -82,7 +82,7 @@\n     return dependency_pair.where(pg_identify_object.c.type == 'table constraint')\n \n \n-# getting a full list of dependents and identifying them\n+# stmt for getting a full list of dependents and identifying them\n def _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject):\n     result = select(\n         pg_depend,\n@@ -136,8 +136,8 @@\n \n     # each statement filters the base statement extracting dependents of a specific type\n     # so it's easy to exclude particular types or add new\n-    dependency_pair = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)\n-    foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair).cte('foreign_key_constraint_dependents')\n+    dependency_pairs = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)\n+    foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pairs).cte('foreign_key_constraint_dependents')\n     table_dependents = _get_table_dependents(foreign_key_constraint_dependents, pg_constraint).cte('table_dependents')\n \n     return union(\n", "issue": "Refactor dependents tests to use library usecase\n## Problem\r\nInitially, both DB and API tests were created using dummy data and were quite difficult to read. \r\n\r\n## Proposed solution\r\nRefactor them to use a common use case and simplify the cases.\r\n\n", "before_files": [{"content": "from sqlalchemy import MetaData, Table, any_, column, exists, func, literal, select, text, true, union\nfrom sqlalchemy.dialects.postgresql import array\n\n# OIDs assigned during normal database operation are constrained to be 16384 or higher.\nUSER_DEFINED_OBJECTS_MIN_OID = 16384\n# automatic and normal dependents\nPG_DEPENDENT_TYPES = ['a', 'n']\nPG_CLASS_CATALOGUE_NAME = '\\'pg_class\\''\nSTART_LEVEL = 1\nMAX_LEVEL = 10\n\n\ndef get_dependents_graph(referenced_object_id, engine):\n    dependency_pairs = _get_typed_dependency_pairs_stmt(engine)\n    dependency_pairs_cte = dependency_pairs.cte(recursive=True, name='dependency_pairs_cte')\n\n    # anchor member which includes all dependents of a requested object\n    anchor = select(\n        dependency_pairs_cte,\n        literal(START_LEVEL).label('level'),\n        array([dependency_pairs_cte.c.refobjid]).label('dependency_chain')) \\\n        .where(dependency_pairs_cte.c.refobjid == referenced_object_id) \\\n        .where(dependency_pairs_cte.c.objid != referenced_object_id)\n    anchor = anchor.cte('cte')\n\n    # recursive member which includes dependents for each object of the previous level\n    recursive = select(\n        dependency_pairs_cte,\n        (anchor.c.level + 1).label('level'),\n        anchor.c.dependency_chain + array([anchor.c.objid])) \\\n        .where(anchor.c.level < MAX_LEVEL) \\\n        .where(dependency_pairs_cte.c.objid != any_(anchor.c.dependency_chain)) \\\n        .where(dependency_pairs_cte.c.objid != dependency_pairs_cte.c.refobjid)\n    recursive = recursive.join(anchor, dependency_pairs_cte.c.refobjid == anchor.c.objid)\n\n    recursive_stmt = anchor.union(recursive)\n    stmt = select(recursive_stmt)\n\n    with engine.connect() as conn:\n        result = conn.execute(stmt)\n\n    return _get_structured_result(result)\n\n\n# finding table dependents based on foreign key constraints from the referenced tables\ndef _get_table_dependents(foreign_key_dependents, pg_constraint_table):\n    # TODO: update refobjsubid with actual values when working on columns\n    pg_identify_object = _get_pg_identify_object_lateral_stmt(\n        text(f'{PG_CLASS_CATALOGUE_NAME}::regclass::oid'), pg_constraint_table.c.conrelid, 0)\n\n    pg_identify_refobject = _get_pg_identify_object_lateral_stmt(\n        foreign_key_dependents.c.refclassid, foreign_key_dependents.c.refobjid, 0)\n\n    # conrelid in this case is the oid of the table which a constraint resides in\n    return select(\n        foreign_key_dependents.c.classid,\n        pg_constraint_table.c.conrelid.label('objid'),\n        foreign_key_dependents.c.objsubid,\n        foreign_key_dependents.c.refclassid,\n        foreign_key_dependents.c.refobjid,\n        foreign_key_dependents.c.refobjsubid,\n        foreign_key_dependents.c.deptype,\n        pg_identify_object.c.name.label('objname'),\n        pg_identify_object.c.type.label('objtype'),\n        pg_identify_refobject.c.name.label('refobjname'),\n        pg_identify_refobject.c.type.label('refobjtype')) \\\n        .select_from(foreign_key_dependents) \\\n        .join(pg_constraint_table, pg_constraint_table.c.oid == foreign_key_dependents.c.objid) \\\n        .join(pg_identify_object, true()) \\\n        .join(pg_identify_refobject, true()) \\\n        .where(pg_constraint_table.c.confrelid != 0) \\\n        .group_by(\n            foreign_key_dependents,\n            pg_constraint_table.c.conrelid,\n            pg_identify_object.c.name,\n            pg_identify_object.c.type,\n            pg_identify_refobject.c.name,\n            pg_identify_refobject.c.type)\n\n\ndef _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair):\n    return dependency_pair.where(pg_identify_object.c.type == 'table constraint')\n\n\n# getting a full list of dependents and identifying them\ndef _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject):\n    result = select(\n        pg_depend,\n        pg_identify_object.c.name.label('objname'),\n        pg_identify_object.c.type.label('objtype'),\n        pg_identify_refobject.c.name.label('refobjname'),\n        pg_identify_refobject.c.type.label('refobjtype')) \\\n        .select_from(pg_depend) \\\n        .join(pg_identify_object, true()) \\\n        .join(pg_identify_refobject, true()) \\\n        .where(pg_depend.c.deptype == any_(array(PG_DEPENDENT_TYPES))) \\\n        .where(pg_depend.c.objid >= USER_DEFINED_OBJECTS_MIN_OID) \\\n        .group_by(\n            pg_depend,\n            pg_identify_object.c.name,\n            pg_identify_object.c.type,\n            pg_identify_refobject.c.name,\n            pg_identify_refobject.c.type)\n\n    return result\n\n\ndef _get_pg_depend_table(engine, metadata):\n    return Table(\"pg_depend\", metadata, autoload_with=engine)\n\n\ndef _get_pg_constraint_table(engine, metadata):\n    return Table(\"pg_constraint\", metadata, autoload_with=engine)\n\n\ndef _get_pg_identify_object_lateral_stmt(classid, objid, objsubid):\n    return select(\n        column(\"name\"),\n        column(\"type\")) \\\n        .select_from(func.pg_identify_object(\n            classid,\n            objid,\n            objsubid)) \\\n        .lateral()\n\n\ndef _get_typed_dependency_pairs_stmt(engine):\n    metadata = MetaData()\n\n    pg_depend = _get_pg_depend_table(engine, metadata)\n    pg_identify_object = _get_pg_identify_object_lateral_stmt(\n        pg_depend.c.classid, pg_depend.c.objid, pg_depend.c.objsubid)\n    pg_identify_refobject = _get_pg_identify_object_lateral_stmt(\n        pg_depend.c.refclassid, pg_depend.c.refobjid, 0)\n    pg_constraint = _get_pg_constraint_table(engine, metadata)\n\n    # each statement filters the base statement extracting dependents of a specific type\n    # so it's easy to exclude particular types or add new\n    dependency_pair = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)\n    foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair).cte('foreign_key_constraint_dependents')\n    table_dependents = _get_table_dependents(foreign_key_constraint_dependents, pg_constraint).cte('table_dependents')\n\n    return union(\n        select(foreign_key_constraint_dependents),\n        select(table_dependents))\n\n\ndef has_dependencies(referenced_object_id, engine):\n    metadata = MetaData()\n\n    pg_depend = _get_pg_depend_table(engine, metadata)\n\n    stmt = select(\n        exists(\n            select().select_from(pg_depend)\n            .where(pg_depend.c.refobjid == referenced_object_id)\n            .where(pg_depend.c.deptype == any_(array(PG_DEPENDENT_TYPES)))\n            .where(pg_depend.c.objid >= USER_DEFINED_OBJECTS_MIN_OID)\n        )\n    )\n\n    with engine.connect() as conn:\n        result = conn.execute(stmt).scalar()\n\n    return result\n\n\ndef _get_structured_result(dependency_graph_result):\n    result = []\n    for dependency_pair in dependency_graph_result:\n        d = {}\n        d['level'] = dependency_pair.level\n        d['obj'] = {'objid': dependency_pair.objid, 'type': dependency_pair.objtype}\n        d['parent_obj'] = {'objid': dependency_pair.refobjid, 'type': dependency_pair.refobjtype}\n        result.append(d)\n\n    return result\n", "path": "db/dependents/dependents_utils.py"}], "after_files": [{"content": "from sqlalchemy import MetaData, Table, any_, column, exists, func, literal, select, text, true, union\nfrom sqlalchemy.dialects.postgresql import array\n\n# OIDs assigned during normal database operation are constrained to be 16384 or higher.\nUSER_DEFINED_OBJECTS_MIN_OID = 16384\n# automatic and normal dependents\nPG_DEPENDENT_TYPES = ['a', 'n']\nPG_CLASS_CATALOGUE_NAME = '\\'pg_class\\''\nSTART_LEVEL = 1\nMAX_LEVEL = 10\n\n\ndef get_dependents_graph(referenced_object_id, engine):\n    dependency_pairs = _get_typed_dependency_pairs_stmt(engine)\n    dependency_pairs_cte = dependency_pairs.cte(recursive=True, name='dependency_pairs_cte')\n\n    # anchor member which includes all dependents of a requested object\n    anchor = select(\n        dependency_pairs_cte,\n        literal(START_LEVEL).label('level'),\n        array([dependency_pairs_cte.c.refobjid]).label('dependency_chain')) \\\n        .where(dependency_pairs_cte.c.refobjid == referenced_object_id) \\\n        .where(dependency_pairs_cte.c.objid != referenced_object_id)\n    anchor = anchor.cte('cte')\n\n    # recursive member which includes dependents for each object of the previous level\n    recursive = select(\n        dependency_pairs_cte,\n        (anchor.c.level + 1),\n        anchor.c.dependency_chain + array([anchor.c.objid])) \\\n        .where(anchor.c.level < MAX_LEVEL) \\\n        .where(dependency_pairs_cte.c.objid != any_(anchor.c.dependency_chain)) \\\n        .where(dependency_pairs_cte.c.objid != dependency_pairs_cte.c.refobjid)\n    recursive = recursive.join(anchor, dependency_pairs_cte.c.refobjid == anchor.c.objid)\n\n    recursive_stmt = anchor.union(recursive)\n    stmt = select(recursive_stmt)\n\n    with engine.connect() as conn:\n        result = conn.execute(stmt)\n\n    return _get_structured_result(result)\n\n\n# finding table dependents based on foreign key constraints from the referenced tables\ndef _get_table_dependents(foreign_key_dependents, pg_constraint_table):\n    # TODO: update refobjsubid with actual values when working on columns\n    pg_identify_object = _get_pg_identify_object_lateral_stmt(\n        text(f'{PG_CLASS_CATALOGUE_NAME}::regclass::oid'), pg_constraint_table.c.conrelid, 0)\n\n    pg_identify_refobject = _get_pg_identify_object_lateral_stmt(\n        foreign_key_dependents.c.refclassid, foreign_key_dependents.c.refobjid, 0)\n\n    # conrelid in this case is the oid of the table which a constraint resides in\n    return select(\n        foreign_key_dependents.c.classid,\n        pg_constraint_table.c.conrelid.label('objid'),\n        foreign_key_dependents.c.objsubid,\n        foreign_key_dependents.c.refclassid,\n        foreign_key_dependents.c.refobjid,\n        foreign_key_dependents.c.refobjsubid,\n        foreign_key_dependents.c.deptype,\n        pg_identify_object.c.name.label('objname'),\n        pg_identify_object.c.type.label('objtype'),\n        pg_identify_refobject.c.name.label('refobjname'),\n        pg_identify_refobject.c.type.label('refobjtype')) \\\n        .select_from(foreign_key_dependents) \\\n        .join(pg_constraint_table, pg_constraint_table.c.oid == foreign_key_dependents.c.objid) \\\n        .join(pg_identify_object, true()) \\\n        .join(pg_identify_refobject, true()) \\\n        .where(pg_constraint_table.c.confrelid != 0) \\\n        .group_by(\n            foreign_key_dependents,\n            pg_constraint_table.c.conrelid,\n            pg_identify_object.c.name,\n            pg_identify_object.c.type,\n            pg_identify_refobject.c.name,\n            pg_identify_refobject.c.type)\n\n\ndef _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pair):\n    return dependency_pair.where(pg_identify_object.c.type == 'table constraint')\n\n\n# stmt for getting a full list of dependents and identifying them\ndef _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject):\n    result = select(\n        pg_depend,\n        pg_identify_object.c.name.label('objname'),\n        pg_identify_object.c.type.label('objtype'),\n        pg_identify_refobject.c.name.label('refobjname'),\n        pg_identify_refobject.c.type.label('refobjtype')) \\\n        .select_from(pg_depend) \\\n        .join(pg_identify_object, true()) \\\n        .join(pg_identify_refobject, true()) \\\n        .where(pg_depend.c.deptype == any_(array(PG_DEPENDENT_TYPES))) \\\n        .where(pg_depend.c.objid >= USER_DEFINED_OBJECTS_MIN_OID) \\\n        .group_by(\n            pg_depend,\n            pg_identify_object.c.name,\n            pg_identify_object.c.type,\n            pg_identify_refobject.c.name,\n            pg_identify_refobject.c.type)\n\n    return result\n\n\ndef _get_pg_depend_table(engine, metadata):\n    return Table(\"pg_depend\", metadata, autoload_with=engine)\n\n\ndef _get_pg_constraint_table(engine, metadata):\n    return Table(\"pg_constraint\", metadata, autoload_with=engine)\n\n\ndef _get_pg_identify_object_lateral_stmt(classid, objid, objsubid):\n    return select(\n        column(\"name\"),\n        column(\"type\")) \\\n        .select_from(func.pg_identify_object(\n            classid,\n            objid,\n            objsubid)) \\\n        .lateral()\n\n\ndef _get_typed_dependency_pairs_stmt(engine):\n    metadata = MetaData()\n\n    pg_depend = _get_pg_depend_table(engine, metadata)\n    pg_identify_object = _get_pg_identify_object_lateral_stmt(\n        pg_depend.c.classid, pg_depend.c.objid, pg_depend.c.objsubid)\n    pg_identify_refobject = _get_pg_identify_object_lateral_stmt(\n        pg_depend.c.refclassid, pg_depend.c.refobjid, 0)\n    pg_constraint = _get_pg_constraint_table(engine, metadata)\n\n    # each statement filters the base statement extracting dependents of a specific type\n    # so it's easy to exclude particular types or add new\n    dependency_pairs = _get_dependency_pairs_stmt(pg_depend, pg_identify_object, pg_identify_refobject)\n    foreign_key_constraint_dependents = _get_foreign_key_constraint_dependents(pg_identify_object, dependency_pairs).cte('foreign_key_constraint_dependents')\n    table_dependents = _get_table_dependents(foreign_key_constraint_dependents, pg_constraint).cte('table_dependents')\n\n    return union(\n        select(foreign_key_constraint_dependents),\n        select(table_dependents))\n\n\ndef has_dependencies(referenced_object_id, engine):\n    metadata = MetaData()\n\n    pg_depend = _get_pg_depend_table(engine, metadata)\n\n    stmt = select(\n        exists(\n            select().select_from(pg_depend)\n            .where(pg_depend.c.refobjid == referenced_object_id)\n            .where(pg_depend.c.deptype == any_(array(PG_DEPENDENT_TYPES)))\n            .where(pg_depend.c.objid >= USER_DEFINED_OBJECTS_MIN_OID)\n        )\n    )\n\n    with engine.connect() as conn:\n        result = conn.execute(stmt).scalar()\n\n    return result\n\n\ndef _get_structured_result(dependency_graph_result):\n    result = []\n    for dependency_pair in dependency_graph_result:\n        d = {}\n        d['level'] = dependency_pair.level\n        d['obj'] = {'objid': dependency_pair.objid, 'type': dependency_pair.objtype}\n        d['parent_obj'] = {'objid': dependency_pair.refobjid, 'type': dependency_pair.refobjtype}\n        result.append(d)\n\n    return result\n", "path": "db/dependents/dependents_utils.py"}]}