We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Make logging of insert statements for write_to_source optional or just remove it
`cesarsantos#3303` in Discord asked for this probably to avoid logging sensible information or filling the logger with all the data there is.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/faldbt/lib.py`
Content:
```
1 # NOTE: INSPIRED IN https://github.com/dbt-labs/dbt-core/blob/43edc887f97e359b02b6317a9f91898d3d66652b/core/dbt/lib.py
2 import six
3 from datetime import datetime
4 from dataclasses import dataclass
5 from uuid import uuid4
6 from typing import List, Tuple, Union
7 
8 import dbt.version
9 import dbt.semver
10 import dbt.flags as flags
11 import dbt.adapters.factory as adapters_factory
12 from dbt.config.runtime import RuntimeConfig
13 from dbt.contracts.connection import AdapterResponse
14 from dbt.contracts.graph.manifest import Manifest
15 from dbt.parser.manifest import process_node
16 from dbt.logger import GLOBAL_LOGGER as logger
17 
18 from . import parse
19 
20 import pandas as pd
21 from pandas.io import sql as pdsql
22 
23 import sqlalchemy
24 from sqlalchemy.sql.ddl import CreateTable
25 from sqlalchemy.sql import Insert
26 from sqlalchemy.sql.schema import MetaData
27 
28 
29 DBT_V1 = dbt.semver.VersionSpecifier.from_version_string("1.0.0")
30 DBT_VCURRENT = dbt.version.get_installed_version()
31 
32 if DBT_VCURRENT.compare(DBT_V1) >= 0:
33     from dbt.parser.sql import SqlBlockParser
34     from dbt.contracts.graph.parsed import ParsedModelNode, ParsedSourceDefinition
35     from dbt.contracts.sql import ResultTable, RemoteRunResult
36 else:
37     from faldbt.cp.parser.sql import SqlBlockParser
38     from faldbt.cp.contracts.graph.parsed import ParsedModelNode, ParsedSourceDefinition
39     from faldbt.cp.contracts.sql import ResultTable, RemoteRunResult
40 
41 
42 @dataclass
43 class FlagsArgs:
44     profiles_dir: str
45     use_colors: bool
46 
47 
48 def initialize_dbt_flags(profiles_dir: str):
49     """
50     Initializes the flags module from dbt, since it's accessed from around their code.
51     """
52     args = FlagsArgs(profiles_dir, None)
53     user_config = parse.get_dbt_user_config(profiles_dir)
54     try:
55         flags.set_from_args(args, user_config)
56     except TypeError:
57         flags.set_from_args(args)
58 
59     # Set invocation id
60     if DBT_VCURRENT.compare(DBT_V1) >= 0:
61         import dbt.events.functions as events_functions
62 
63         events_functions.set_invocation_id()
64 
65     # Re-enable logging for 1.0.0 through old API of logger
66     # TODO: migrate for 1.0.0 code to new event system
67     if DBT_VCURRENT.compare(DBT_V1) >= 0:
68         flags.ENABLE_LEGACY_LOGGER = "1"
69         if logger.disabled:
70             logger.enable()
71 
72 
73 def register_adapters(config: RuntimeConfig):
74     # Clear previously registered adapters. This fixes cacheing behavior on the dbt-server
75     adapters_factory.reset_adapters()
76     # Load the relevant adapter
77     adapters_factory.register_adapter(config)
78 
79 
80 def _get_operation_node(manifest: Manifest, project_path, profiles_dir, sql):
81 
82     config = parse.get_dbt_config(project_path, profiles_dir)
83     block_parser = SqlBlockParser(
84         project=config,
85         manifest=manifest,
86         root_project=config,
87     )
88 
89     # NOTE: nodes get registered to the manifest automatically,
90     # HACK: we need to include uniqueness (UUID4) to avoid clashes
91     name = "SQL:" + str(hash(sql)) + ":" + str(uuid4())
92     sql_node = block_parser.parse_remote(sql, name)
93     process_node(config, manifest, sql_node)
94     return sql_node
95 
96 
97 # NOTE: Once we get an adapter, we must call `connection_for` or `connection_named` to use it
98 def _get_adapter(project_path: str, profiles_dir: str):
99     config = parse.get_dbt_config(project_path, profiles_dir)
100 
101     adapters_factory.cleanup_connections()
102     return adapters_factory.get_adapter(config)
103 
104 
105 def _execute_sql(
106     manifest: Manifest, project_path: str, profiles_dir: str, sql: str
107 ) -> Tuple[AdapterResponse, RemoteRunResult]:
108     node = _get_operation_node(manifest, project_path, profiles_dir, sql)
109     adapter = _get_adapter(project_path, profiles_dir)
110 
111     logger.info("Running query\n{}", sql)
112 
113     result = None
114     with adapter.connection_for(node):
115         adapter.connections.begin()
116         response, execute_result = adapter.execute(sql, fetch=True)
117 
118         table = ResultTable(
119             column_names=list(execute_result.column_names),
120             rows=[list(row) for row in execute_result],
121         )
122 
123         result = RemoteRunResult(
124             raw_sql=sql,
125             compiled_sql=sql,
126             node=node,
127             table=table,
128             timing=[],
129             logs=[],
130             generated_at=datetime.utcnow(),
131         )
132         adapter.connections.commit()
133 
134     return response, result
135 
136 
137 def _get_target_relation(
138     target: Union[ParsedModelNode, ParsedSourceDefinition],
139     project_path: str,
140     profiles_dir: str,
141 ):
142     adapter = _get_adapter(project_path, profiles_dir)
143 
144     relation = None
145     with adapter.connection_named(str(uuid4())):
146         # This ROLLBACKs so it has to be a new connection
147         relation = adapter.get_relation(
148             target.database, target.schema, target.identifier
149         )
150     return relation
151 
152 
153 def execute_sql(
154     manifest: Manifest, project_path: str, profiles_dir: str, sql: str
155 ) -> RemoteRunResult:
156     _, result = _execute_sql(manifest, project_path, profiles_dir, sql)
157     return result
158 
159 
160 def fetch_target(
161     manifest: Manifest,
162     project_path: str,
163     profiles_dir: str,
164     target: Union[ParsedModelNode, ParsedSourceDefinition],
165 ) -> RemoteRunResult:
166     relation = _get_target_relation(target, project_path, profiles_dir)
167 
168     if relation is None:
169         raise Exception(f"Could not get relation for '{target.unique_id}'")
170 
171     query = f"SELECT * FROM {relation}"
172     _, result = _execute_sql(manifest, project_path, profiles_dir, query)
173     return result
174 
175 
176 def write_target(
177     data: pd.DataFrame,
178     manifest: Manifest,
179     project_path: str,
180     profiles_dir: str,
181     target: Union[ParsedModelNode, ParsedSourceDefinition],
182 ) -> RemoteRunResult:
183     adapter = _get_adapter(project_path, profiles_dir)
184 
185     relation = _get_target_relation(target, project_path, profiles_dir)
186 
187     engine = _alchemy_engine(adapter, target)
188     pddb = pdsql.SQLDatabase(engine, schema=target.schema)
189     pdtable = pdsql.SQLTable(target.name, pddb, data, index=False)
190     alchemy_table: sqlalchemy.Table = pdtable.table.to_metadata(pdtable.pd_sql.meta)
191 
192     column_names: List[str] = list(data.columns)
193     rows = data.to_records(index=False)
194     row_dicts = list(map(lambda row: dict(zip(column_names, row)), rows))
195 
196     if relation is None:
197         create_stmt = CreateTable(alchemy_table).compile(
198             bind=engine, compile_kwargs={"literal_binds": True}
199         )
200 
201         _execute_sql(
202             manifest, project_path, profiles_dir, six.text_type(create_stmt).strip()
203         )
204 
205     insert_stmt = (
206         Insert(alchemy_table)
207         .values(row_dicts)
208         .compile(bind=engine, compile_kwargs={"literal_binds": True})
209     )
210 
211     _, result = _execute_sql(
212         manifest, project_path, profiles_dir, six.text_type(insert_stmt).strip()
213     )
214     return result
215 
216 
217 def _alchemy_engine(
218     adapter: adapters_factory.Adapter,
219     target: Union[ParsedModelNode, ParsedSourceDefinition],
220 ):
221     url_string = f"{adapter.type()}://"
222     if adapter.type() == "bigquery":
223         url_string = f"bigquery://{target.database}"
224     if adapter.type() == "postgres":
225         url_string = "postgresql://"
226     # TODO: add special cases as needed
227 
228     def null_dump(sql, *multiparams, **params):
229         pass
230 
231     return sqlalchemy.create_mock_engine(url_string, executor=null_dump)
232
```
--- 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/src/faldbt/lib.py b/src/faldbt/lib.py
--- a/src/faldbt/lib.py
+++ b/src/faldbt/lib.py
@@ -108,7 +108,7 @@
     node = _get_operation_node(manifest, project_path, profiles_dir, sql)
     adapter = _get_adapter(project_path, profiles_dir)
 
-    logger.info("Running query\n{}", sql)
+    logger.debug("Running query\n{}", sql)
 
     result = None
     with adapter.connection_for(node):

{"golden_diff": "diff --git a/src/faldbt/lib.py b/src/faldbt/lib.py\n--- a/src/faldbt/lib.py\n+++ b/src/faldbt/lib.py\n@@ -108,7 +108,7 @@\n     node = _get_operation_node(manifest, project_path, profiles_dir, sql)\n     adapter = _get_adapter(project_path, profiles_dir)\n \n-    logger.info(\"Running query\\n{}\", sql)\n+    logger.debug(\"Running query\\n{}\", sql)\n \n     result = None\n     with adapter.connection_for(node):\n", "issue": "Make logging of insert statements for write_to_source optional or just remove it\n`cesarsantos#3303` in Discord asked for this probably to avoid logging sensible information or filling the logger with all the data there is.\n", "before_files": [{"content": "# NOTE: INSPIRED IN https://github.com/dbt-labs/dbt-core/blob/43edc887f97e359b02b6317a9f91898d3d66652b/core/dbt/lib.py\nimport six\nfrom datetime import datetime\nfrom dataclasses import dataclass\nfrom uuid import uuid4\nfrom typing import List, Tuple, Union\n\nimport dbt.version\nimport dbt.semver\nimport dbt.flags as flags\nimport dbt.adapters.factory as adapters_factory\nfrom dbt.config.runtime import RuntimeConfig\nfrom dbt.contracts.connection import AdapterResponse\nfrom dbt.contracts.graph.manifest import Manifest\nfrom dbt.parser.manifest import process_node\nfrom dbt.logger import GLOBAL_LOGGER as logger\n\nfrom . import parse\n\nimport pandas as pd\nfrom pandas.io import sql as pdsql\n\nimport sqlalchemy\nfrom sqlalchemy.sql.ddl import CreateTable\nfrom sqlalchemy.sql import Insert\nfrom sqlalchemy.sql.schema import MetaData\n\n\nDBT_V1 = dbt.semver.VersionSpecifier.from_version_string(\"1.0.0\")\nDBT_VCURRENT = dbt.version.get_installed_version()\n\nif DBT_VCURRENT.compare(DBT_V1) >= 0:\n    from dbt.parser.sql import SqlBlockParser\n    from dbt.contracts.graph.parsed import ParsedModelNode, ParsedSourceDefinition\n    from dbt.contracts.sql import ResultTable, RemoteRunResult\nelse:\n    from faldbt.cp.parser.sql import SqlBlockParser\n    from faldbt.cp.contracts.graph.parsed import ParsedModelNode, ParsedSourceDefinition\n    from faldbt.cp.contracts.sql import ResultTable, RemoteRunResult\n\n\n@dataclass\nclass FlagsArgs:\n    profiles_dir: str\n    use_colors: bool\n\n\ndef initialize_dbt_flags(profiles_dir: str):\n    \"\"\"\n    Initializes the flags module from dbt, since it's accessed from around their code.\n    \"\"\"\n    args = FlagsArgs(profiles_dir, None)\n    user_config = parse.get_dbt_user_config(profiles_dir)\n    try:\n        flags.set_from_args(args, user_config)\n    except TypeError:\n        flags.set_from_args(args)\n\n    # Set invocation id\n    if DBT_VCURRENT.compare(DBT_V1) >= 0:\n        import dbt.events.functions as events_functions\n\n        events_functions.set_invocation_id()\n\n    # Re-enable logging for 1.0.0 through old API of logger\n    # TODO: migrate for 1.0.0 code to new event system\n    if DBT_VCURRENT.compare(DBT_V1) >= 0:\n        flags.ENABLE_LEGACY_LOGGER = \"1\"\n        if logger.disabled:\n            logger.enable()\n\n\ndef register_adapters(config: RuntimeConfig):\n    # Clear previously registered adapters. This fixes cacheing behavior on the dbt-server\n    adapters_factory.reset_adapters()\n    # Load the relevant adapter\n    adapters_factory.register_adapter(config)\n\n\ndef _get_operation_node(manifest: Manifest, project_path, profiles_dir, sql):\n\n    config = parse.get_dbt_config(project_path, profiles_dir)\n    block_parser = SqlBlockParser(\n        project=config,\n        manifest=manifest,\n        root_project=config,\n    )\n\n    # NOTE: nodes get registered to the manifest automatically,\n    # HACK: we need to include uniqueness (UUID4) to avoid clashes\n    name = \"SQL:\" + str(hash(sql)) + \":\" + str(uuid4())\n    sql_node = block_parser.parse_remote(sql, name)\n    process_node(config, manifest, sql_node)\n    return sql_node\n\n\n# NOTE: Once we get an adapter, we must call `connection_for` or `connection_named` to use it\ndef _get_adapter(project_path: str, profiles_dir: str):\n    config = parse.get_dbt_config(project_path, profiles_dir)\n\n    adapters_factory.cleanup_connections()\n    return adapters_factory.get_adapter(config)\n\n\ndef _execute_sql(\n    manifest: Manifest, project_path: str, profiles_dir: str, sql: str\n) -> Tuple[AdapterResponse, RemoteRunResult]:\n    node = _get_operation_node(manifest, project_path, profiles_dir, sql)\n    adapter = _get_adapter(project_path, profiles_dir)\n\n    logger.info(\"Running query\\n{}\", sql)\n\n    result = None\n    with adapter.connection_for(node):\n        adapter.connections.begin()\n        response, execute_result = adapter.execute(sql, fetch=True)\n\n        table = ResultTable(\n            column_names=list(execute_result.column_names),\n            rows=[list(row) for row in execute_result],\n        )\n\n        result = RemoteRunResult(\n            raw_sql=sql,\n            compiled_sql=sql,\n            node=node,\n            table=table,\n            timing=[],\n            logs=[],\n            generated_at=datetime.utcnow(),\n        )\n        adapter.connections.commit()\n\n    return response, result\n\n\ndef _get_target_relation(\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n    project_path: str,\n    profiles_dir: str,\n):\n    adapter = _get_adapter(project_path, profiles_dir)\n\n    relation = None\n    with adapter.connection_named(str(uuid4())):\n        # This ROLLBACKs so it has to be a new connection\n        relation = adapter.get_relation(\n            target.database, target.schema, target.identifier\n        )\n    return relation\n\n\ndef execute_sql(\n    manifest: Manifest, project_path: str, profiles_dir: str, sql: str\n) -> RemoteRunResult:\n    _, result = _execute_sql(manifest, project_path, profiles_dir, sql)\n    return result\n\n\ndef fetch_target(\n    manifest: Manifest,\n    project_path: str,\n    profiles_dir: str,\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n) -> RemoteRunResult:\n    relation = _get_target_relation(target, project_path, profiles_dir)\n\n    if relation is None:\n        raise Exception(f\"Could not get relation for '{target.unique_id}'\")\n\n    query = f\"SELECT * FROM {relation}\"\n    _, result = _execute_sql(manifest, project_path, profiles_dir, query)\n    return result\n\n\ndef write_target(\n    data: pd.DataFrame,\n    manifest: Manifest,\n    project_path: str,\n    profiles_dir: str,\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n) -> RemoteRunResult:\n    adapter = _get_adapter(project_path, profiles_dir)\n\n    relation = _get_target_relation(target, project_path, profiles_dir)\n\n    engine = _alchemy_engine(adapter, target)\n    pddb = pdsql.SQLDatabase(engine, schema=target.schema)\n    pdtable = pdsql.SQLTable(target.name, pddb, data, index=False)\n    alchemy_table: sqlalchemy.Table = pdtable.table.to_metadata(pdtable.pd_sql.meta)\n\n    column_names: List[str] = list(data.columns)\n    rows = data.to_records(index=False)\n    row_dicts = list(map(lambda row: dict(zip(column_names, row)), rows))\n\n    if relation is None:\n        create_stmt = CreateTable(alchemy_table).compile(\n            bind=engine, compile_kwargs={\"literal_binds\": True}\n        )\n\n        _execute_sql(\n            manifest, project_path, profiles_dir, six.text_type(create_stmt).strip()\n        )\n\n    insert_stmt = (\n        Insert(alchemy_table)\n        .values(row_dicts)\n        .compile(bind=engine, compile_kwargs={\"literal_binds\": True})\n    )\n\n    _, result = _execute_sql(\n        manifest, project_path, profiles_dir, six.text_type(insert_stmt).strip()\n    )\n    return result\n\n\ndef _alchemy_engine(\n    adapter: adapters_factory.Adapter,\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n):\n    url_string = f\"{adapter.type()}://\"\n    if adapter.type() == \"bigquery\":\n        url_string = f\"bigquery://{target.database}\"\n    if adapter.type() == \"postgres\":\n        url_string = \"postgresql://\"\n    # TODO: add special cases as needed\n\n    def null_dump(sql, *multiparams, **params):\n        pass\n\n    return sqlalchemy.create_mock_engine(url_string, executor=null_dump)\n", "path": "src/faldbt/lib.py"}], "after_files": [{"content": "# NOTE: INSPIRED IN https://github.com/dbt-labs/dbt-core/blob/43edc887f97e359b02b6317a9f91898d3d66652b/core/dbt/lib.py\nimport six\nfrom datetime import datetime\nfrom dataclasses import dataclass\nfrom uuid import uuid4\nfrom typing import List, Tuple, Union\n\nimport dbt.version\nimport dbt.semver\nimport dbt.flags as flags\nimport dbt.adapters.factory as adapters_factory\nfrom dbt.config.runtime import RuntimeConfig\nfrom dbt.contracts.connection import AdapterResponse\nfrom dbt.contracts.graph.manifest import Manifest\nfrom dbt.parser.manifest import process_node\nfrom dbt.logger import GLOBAL_LOGGER as logger\n\nfrom . import parse\n\nimport pandas as pd\nfrom pandas.io import sql as pdsql\n\nimport sqlalchemy\nfrom sqlalchemy.sql.ddl import CreateTable\nfrom sqlalchemy.sql import Insert\nfrom sqlalchemy.sql.schema import MetaData\n\n\nDBT_V1 = dbt.semver.VersionSpecifier.from_version_string(\"1.0.0\")\nDBT_VCURRENT = dbt.version.get_installed_version()\n\nif DBT_VCURRENT.compare(DBT_V1) >= 0:\n    from dbt.parser.sql import SqlBlockParser\n    from dbt.contracts.graph.parsed import ParsedModelNode, ParsedSourceDefinition\n    from dbt.contracts.sql import ResultTable, RemoteRunResult\nelse:\n    from faldbt.cp.parser.sql import SqlBlockParser\n    from faldbt.cp.contracts.graph.parsed import ParsedModelNode, ParsedSourceDefinition\n    from faldbt.cp.contracts.sql import ResultTable, RemoteRunResult\n\n\n@dataclass\nclass FlagsArgs:\n    profiles_dir: str\n    use_colors: bool\n\n\ndef initialize_dbt_flags(profiles_dir: str):\n    \"\"\"\n    Initializes the flags module from dbt, since it's accessed from around their code.\n    \"\"\"\n    args = FlagsArgs(profiles_dir, None)\n    user_config = parse.get_dbt_user_config(profiles_dir)\n    try:\n        flags.set_from_args(args, user_config)\n    except TypeError:\n        flags.set_from_args(args)\n\n    # Set invocation id\n    if DBT_VCURRENT.compare(DBT_V1) >= 0:\n        import dbt.events.functions as events_functions\n\n        events_functions.set_invocation_id()\n\n    # Re-enable logging for 1.0.0 through old API of logger\n    # TODO: migrate for 1.0.0 code to new event system\n    if DBT_VCURRENT.compare(DBT_V1) >= 0:\n        flags.ENABLE_LEGACY_LOGGER = \"1\"\n        if logger.disabled:\n            logger.enable()\n\n\ndef register_adapters(config: RuntimeConfig):\n    # Clear previously registered adapters. This fixes cacheing behavior on the dbt-server\n    adapters_factory.reset_adapters()\n    # Load the relevant adapter\n    adapters_factory.register_adapter(config)\n\n\ndef _get_operation_node(manifest: Manifest, project_path, profiles_dir, sql):\n\n    config = parse.get_dbt_config(project_path, profiles_dir)\n    block_parser = SqlBlockParser(\n        project=config,\n        manifest=manifest,\n        root_project=config,\n    )\n\n    # NOTE: nodes get registered to the manifest automatically,\n    # HACK: we need to include uniqueness (UUID4) to avoid clashes\n    name = \"SQL:\" + str(hash(sql)) + \":\" + str(uuid4())\n    sql_node = block_parser.parse_remote(sql, name)\n    process_node(config, manifest, sql_node)\n    return sql_node\n\n\n# NOTE: Once we get an adapter, we must call `connection_for` or `connection_named` to use it\ndef _get_adapter(project_path: str, profiles_dir: str):\n    config = parse.get_dbt_config(project_path, profiles_dir)\n\n    adapters_factory.cleanup_connections()\n    return adapters_factory.get_adapter(config)\n\n\ndef _execute_sql(\n    manifest: Manifest, project_path: str, profiles_dir: str, sql: str\n) -> Tuple[AdapterResponse, RemoteRunResult]:\n    node = _get_operation_node(manifest, project_path, profiles_dir, sql)\n    adapter = _get_adapter(project_path, profiles_dir)\n\n    logger.debug(\"Running query\\n{}\", sql)\n\n    result = None\n    with adapter.connection_for(node):\n        adapter.connections.begin()\n        response, execute_result = adapter.execute(sql, fetch=True)\n\n        table = ResultTable(\n            column_names=list(execute_result.column_names),\n            rows=[list(row) for row in execute_result],\n        )\n\n        result = RemoteRunResult(\n            raw_sql=sql,\n            compiled_sql=sql,\n            node=node,\n            table=table,\n            timing=[],\n            logs=[],\n            generated_at=datetime.utcnow(),\n        )\n        adapter.connections.commit()\n\n    return response, result\n\n\ndef _get_target_relation(\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n    project_path: str,\n    profiles_dir: str,\n):\n    adapter = _get_adapter(project_path, profiles_dir)\n\n    relation = None\n    with adapter.connection_named(str(uuid4())):\n        # This ROLLBACKs so it has to be a new connection\n        relation = adapter.get_relation(\n            target.database, target.schema, target.identifier\n        )\n    return relation\n\n\ndef execute_sql(\n    manifest: Manifest, project_path: str, profiles_dir: str, sql: str\n) -> RemoteRunResult:\n    _, result = _execute_sql(manifest, project_path, profiles_dir, sql)\n    return result\n\n\ndef fetch_target(\n    manifest: Manifest,\n    project_path: str,\n    profiles_dir: str,\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n) -> RemoteRunResult:\n    relation = _get_target_relation(target, project_path, profiles_dir)\n\n    if relation is None:\n        raise Exception(f\"Could not get relation for '{target.unique_id}'\")\n\n    query = f\"SELECT * FROM {relation}\"\n    _, result = _execute_sql(manifest, project_path, profiles_dir, query)\n    return result\n\n\ndef write_target(\n    data: pd.DataFrame,\n    manifest: Manifest,\n    project_path: str,\n    profiles_dir: str,\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n) -> RemoteRunResult:\n    adapter = _get_adapter(project_path, profiles_dir)\n\n    relation = _get_target_relation(target, project_path, profiles_dir)\n\n    engine = _alchemy_engine(adapter, target)\n    pddb = pdsql.SQLDatabase(engine, schema=target.schema)\n    pdtable = pdsql.SQLTable(target.name, pddb, data, index=False)\n    alchemy_table: sqlalchemy.Table = pdtable.table.to_metadata(pdtable.pd_sql.meta)\n\n    column_names: List[str] = list(data.columns)\n    rows = data.to_records(index=False)\n    row_dicts = list(map(lambda row: dict(zip(column_names, row)), rows))\n\n    if relation is None:\n        create_stmt = CreateTable(alchemy_table).compile(\n            bind=engine, compile_kwargs={\"literal_binds\": True}\n        )\n\n        _execute_sql(\n            manifest, project_path, profiles_dir, six.text_type(create_stmt).strip()\n        )\n\n    insert_stmt = (\n        Insert(alchemy_table)\n        .values(row_dicts)\n        .compile(bind=engine, compile_kwargs={\"literal_binds\": True})\n    )\n\n    _, result = _execute_sql(\n        manifest, project_path, profiles_dir, six.text_type(insert_stmt).strip()\n    )\n    return result\n\n\ndef _alchemy_engine(\n    adapter: adapters_factory.Adapter,\n    target: Union[ParsedModelNode, ParsedSourceDefinition],\n):\n    url_string = f\"{adapter.type()}://\"\n    if adapter.type() == \"bigquery\":\n        url_string = f\"bigquery://{target.database}\"\n    if adapter.type() == \"postgres\":\n        url_string = \"postgresql://\"\n    # TODO: add special cases as needed\n\n    def null_dump(sql, *multiparams, **params):\n        pass\n\n    return sqlalchemy.create_mock_engine(url_string, executor=null_dump)\n", "path": "src/faldbt/lib.py"}]}