We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Implement .insert() for SQLite
Some backends implement the `.insert()` method, to insert rows into tables. SQLite doesn't, and besides being useful for users, it'd be good to have it working, and add an example in the tutorial 5, so it's useful as a reference for other backends too.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ibis/backends/base/sql/alchemy/client.py`
Content:
```
1 import contextlib
2 import functools
3 from typing import List, Optional
4 
5 import pandas as pd
6 import sqlalchemy as sa
7 from pkg_resources import parse_version
8 
9 import ibis
10 import ibis.expr.datatypes as dt
11 import ibis.expr.schema as sch
12 import ibis.util as util
13 from ibis.backends.base.sql.compiler import Dialect
14 from ibis.client import Query, SQLClient
15 
16 from .datatypes import to_sqla_type
17 from .geospatial import geospatial_supported
18 from .query_builder import build_ast
19 from .translator import AlchemyExprTranslator
20 
21 if geospatial_supported:
22     import geoalchemy2.shape as shape
23     import geopandas
24 
25 
26 class _AlchemyProxy:
27     """
28     Wraps a SQLAlchemy ResultProxy and ensures that .close() is called on
29     garbage collection
30     """
31 
32     def __init__(self, proxy):
33         self.proxy = proxy
34 
35     def __del__(self):
36         self._close_cursor()
37 
38     def _close_cursor(self):
39         self.proxy.close()
40 
41     def __enter__(self):
42         return self
43 
44     def __exit__(self, type, value, tb):
45         self._close_cursor()
46 
47     def fetchall(self):
48         return self.proxy.fetchall()
49 
50 
51 def _invalidates_reflection_cache(f):
52     """Invalidate the SQLAlchemy reflection cache if `f` performs an operation
53     that mutates database or table metadata such as ``CREATE TABLE``,
54     ``DROP TABLE``, etc.
55 
56     Parameters
57     ----------
58     f : callable
59         A method on :class:`ibis.sql.alchemy.AlchemyClient`
60     """
61 
62     @functools.wraps(f)
63     def wrapped(self, *args, **kwargs):
64         result = f(self, *args, **kwargs)
65 
66         # only invalidate the cache after we've succesfully called the wrapped
67         # function
68         self._reflection_cache_is_dirty = True
69         return result
70 
71     return wrapped
72 
73 
74 def _maybe_to_geodataframe(df, schema):
75     """
76     If the required libraries for geospatial support are installed, and if a
77     geospatial column is present in the dataframe, convert it to a
78     GeoDataFrame.
79     """
80 
81     def to_shapely(row, name):
82         return shape.to_shape(row[name]) if row[name] is not None else None
83 
84     if len(df) and geospatial_supported:
85         geom_col = None
86         for name, dtype in schema.items():
87             if isinstance(dtype, dt.GeoSpatial):
88                 geom_col = geom_col or name
89                 df[name] = df.apply(lambda x: to_shapely(x, name), axis=1)
90         if geom_col:
91             df = geopandas.GeoDataFrame(df, geometry=geom_col)
92     return df
93 
94 
95 class AlchemyQuery(Query):
96     def _fetch(self, cursor):
97         df = pd.DataFrame.from_records(
98             cursor.proxy.fetchall(),
99             columns=cursor.proxy.keys(),
100             coerce_float=True,
101         )
102         schema = self.schema()
103         return _maybe_to_geodataframe(schema.apply_to(df), schema)
104 
105 
106 class AlchemyDialect(Dialect):
107 
108     translator = AlchemyExprTranslator
109 
110 
111 class AlchemyClient(SQLClient):
112 
113     dialect = AlchemyDialect
114     query_class = AlchemyQuery
115     has_attachment = False
116 
117     def __init__(self, con: sa.engine.Engine) -> None:
118         super().__init__()
119         self.con = con
120         self.meta = sa.MetaData(bind=con)
121         self._inspector = sa.inspect(con)
122         self._reflection_cache_is_dirty = False
123         self._schemas = {}
124 
125     @property
126     def inspector(self):
127         if self._reflection_cache_is_dirty:
128             self._inspector.info_cache.clear()
129         return self._inspector
130 
131     @contextlib.contextmanager
132     def begin(self):
133         with self.con.begin() as bind:
134             yield bind
135 
136     @_invalidates_reflection_cache
137     def create_table(self, name, expr=None, schema=None, database=None):
138         if database == self.database_name:
139             # avoid fully qualified name
140             database = None
141 
142         if database is not None:
143             raise NotImplementedError(
144                 'Creating tables from a different database is not yet '
145                 'implemented'
146             )
147 
148         if expr is None and schema is None:
149             raise ValueError('You must pass either an expression or a schema')
150 
151         if expr is not None and schema is not None:
152             if not expr.schema().equals(ibis.schema(schema)):
153                 raise TypeError(
154                     'Expression schema is not equal to passed schema. '
155                     'Try passing the expression without the schema'
156                 )
157         if schema is None:
158             schema = expr.schema()
159 
160         self._schemas[self._fully_qualified_name(name, database)] = schema
161         t = self._table_from_schema(
162             name, schema, database=database or self.current_database
163         )
164 
165         with self.begin() as bind:
166             t.create(bind=bind)
167             if expr is not None:
168                 bind.execute(
169                     t.insert().from_select(list(expr.columns), expr.compile())
170                 )
171 
172     def _columns_from_schema(
173         self, name: str, schema: sch.Schema
174     ) -> List[sa.Column]:
175         return [
176             sa.Column(colname, to_sqla_type(dtype), nullable=dtype.nullable)
177             for colname, dtype in zip(schema.names, schema.types)
178         ]
179 
180     def _table_from_schema(
181         self, name: str, schema: sch.Schema, database: Optional[str] = None
182     ) -> sa.Table:
183         columns = self._columns_from_schema(name, schema)
184         return sa.Table(name, self.meta, *columns)
185 
186     @_invalidates_reflection_cache
187     def drop_table(
188         self,
189         table_name: str,
190         database: Optional[str] = None,
191         force: bool = False,
192     ) -> None:
193         if database == self.database_name:
194             # avoid fully qualified name
195             database = None
196 
197         if database is not None:
198             raise NotImplementedError(
199                 'Dropping tables from a different database is not yet '
200                 'implemented'
201             )
202 
203         t = self._get_sqla_table(table_name, schema=database, autoload=False)
204         t.drop(checkfirst=force)
205 
206         assert (
207             not t.exists()
208         ), 'Something went wrong during DROP of table {!r}'.format(t.name)
209 
210         self.meta.remove(t)
211 
212         qualified_name = self._fully_qualified_name(table_name, database)
213 
214         try:
215             del self._schemas[qualified_name]
216         except KeyError:  # schemas won't be cached if created with raw_sql
217             pass
218 
219     def load_data(
220         self,
221         table_name: str,
222         data: pd.DataFrame,
223         database: str = None,
224         if_exists: str = 'fail',
225     ):
226         """
227         Load data from a dataframe to the backend.
228 
229         Parameters
230         ----------
231         table_name : string
232         data: pandas.DataFrame
233         database : string, optional
234         if_exists : string, optional, default 'fail'
235             The values available are: {‘fail’, ‘replace’, ‘append’}
236 
237         Raises
238         ------
239         NotImplementedError
240             Loading data to a table from a different database is not
241             yet implemented
242         """
243         if database == self.database_name:
244             # avoid fully qualified name
245             database = None
246 
247         if database is not None:
248             raise NotImplementedError(
249                 'Loading data to a table from a different database is not '
250                 'yet implemented'
251             )
252 
253         params = {}
254         if self.has_attachment:
255             # for database with attachment
256             # see: https://github.com/ibis-project/ibis/issues/1930
257             params['schema'] = self.database_name
258 
259         data.to_sql(
260             table_name,
261             con=self.con,
262             index=False,
263             if_exists=if_exists,
264             **params,
265         )
266 
267     def truncate_table(
268         self, table_name: str, database: Optional[str] = None
269     ) -> None:
270         t = self._get_sqla_table(table_name, schema=database)
271         t.delete().execute()
272 
273     def list_tables(
274         self,
275         like: Optional[str] = None,
276         database: Optional[str] = None,
277         schema: Optional[str] = None,
278     ) -> List[str]:
279         """List tables/views in the current or indicated database.
280 
281         Parameters
282         ----------
283         like
284             Checks for this string contained in name
285         database
286             If not passed, uses the current database
287         schema
288             The schema namespace that tables should be listed from
289 
290         Returns
291         -------
292         List[str]
293 
294         """
295         inspector = self.inspector
296         # inspector returns a mutable version of its names, so make a copy.
297         names = inspector.get_table_names(schema=schema).copy()
298         names.extend(inspector.get_view_names(schema=schema))
299         if like is not None:
300             names = [x for x in names if like in x]
301         return sorted(names)
302 
303     def _execute(self, query: str, results: bool = True):
304         return _AlchemyProxy(self.con.execute(query))
305 
306     @_invalidates_reflection_cache
307     def raw_sql(self, query: str, results: bool = False):
308         return super().raw_sql(query, results=results)
309 
310     def _build_ast(self, expr, context):
311         return build_ast(expr, context)
312 
313     def _log(self, sql):
314         try:
315             query_str = str(sql)
316         except sa.exc.UnsupportedCompilationError:
317             pass
318         else:
319             util.log(query_str)
320 
321     def _get_sqla_table(self, name, schema=None, autoload=True):
322         return sa.Table(name, self.meta, schema=schema, autoload=autoload)
323 
324     def _sqla_table_to_expr(self, table):
325         node = self.table_class(table, self)
326         return self.table_expr_class(node)
327 
328     @property
329     def version(self):
330         vstring = '.'.join(map(str, self.con.dialect.server_version_info))
331         return parse_version(vstring)
332
```
--- 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/ibis/backends/base/sql/alchemy/client.py b/ibis/backends/base/sql/alchemy/client.py
--- a/ibis/backends/base/sql/alchemy/client.py
+++ b/ibis/backends/base/sql/alchemy/client.py
@@ -1,6 +1,6 @@
 import contextlib
 import functools
-from typing import List, Optional
+from typing import List, Optional, Union
 
 import pandas as pd
 import sqlalchemy as sa
@@ -9,6 +9,7 @@
 import ibis
 import ibis.expr.datatypes as dt
 import ibis.expr.schema as sch
+import ibis.expr.types as ir
 import ibis.util as util
 from ibis.backends.base.sql.compiler import Dialect
 from ibis.client import Query, SQLClient
@@ -229,7 +230,7 @@
         Parameters
         ----------
         table_name : string
-        data: pandas.DataFrame
+        data : pandas.DataFrame
         database : string, optional
         if_exists : string, optional, default 'fail'
             The values available are: {‘fail’, ‘replace’, ‘append’}
@@ -329,3 +330,93 @@
     def version(self):
         vstring = '.'.join(map(str, self.con.dialect.server_version_info))
         return parse_version(vstring)
+
+    def insert(
+        self,
+        table_name: str,
+        obj: Union[pd.DataFrame, ir.TableExpr],
+        database: Optional[str] = None,
+        overwrite: Optional[bool] = False,
+    ) -> None:
+        """
+        Insert the given data to a table in backend.
+
+        Parameters
+        ----------
+        table_name : string
+            name of the table to which data needs to be inserted
+        obj : pandas DataFrame or ibis TableExpr
+            obj is either the dataframe (pd.DataFrame) containing data
+            which needs to be inserted to table_name or
+            the TableExpr type which ibis provides with data which needs
+            to be inserted to table_name
+        database : string, optional
+            name of the attached database that the table is located in.
+        overwrite : boolean, default False
+            If True, will replace existing contents of table else not
+
+        Raises
+        -------
+        NotImplementedError
+            Inserting data to a table from a different database is not
+            yet implemented
+
+        ValueError
+            No operation is being performed. Either the obj parameter
+            is not a pandas DataFrame or is not a ibis TableExpr.
+            The given obj is of type type(obj).__name__ .
+
+        """
+
+        if database == self.database_name:
+            # avoid fully qualified name
+            database = None
+
+        if database is not None:
+            raise NotImplementedError(
+                'Inserting data to a table from a different database is not '
+                'yet implemented'
+            )
+
+        params = {}
+        if self.has_attachment:
+            # for database with attachment
+            # see: https://github.com/ibis-project/ibis/issues/1930
+            params['schema'] = self.database_name
+
+        if isinstance(obj, pd.DataFrame):
+            obj.to_sql(
+                table_name,
+                self.con,
+                index=False,
+                if_exists='replace' if overwrite else 'append',
+                **params,
+            )
+        elif isinstance(obj, ir.TableExpr):
+            to_table_expr = self.table(table_name)
+            to_table_schema = to_table_expr.schema()
+
+            if overwrite:
+                self.drop_table(table_name, database=database)
+                self.create_table(
+                    table_name, schema=to_table_schema, database=database,
+                )
+
+            to_table = self._get_sqla_table(table_name, schema=database)
+
+            from_table_expr = obj
+
+            with self.begin() as bind:
+                if from_table_expr is not None:
+                    bind.execute(
+                        to_table.insert().from_select(
+                            list(from_table_expr.columns),
+                            from_table_expr.compile(),
+                        )
+                    )
+        else:
+            raise ValueError(
+                "No operation is being performed. Either the obj parameter "
+                "is not a pandas DataFrame or is not a ibis TableExpr."
+                f"The given obj is of type {type(obj).__name__} ."
+            )

{"golden_diff": "diff --git a/ibis/backends/base/sql/alchemy/client.py b/ibis/backends/base/sql/alchemy/client.py\n--- a/ibis/backends/base/sql/alchemy/client.py\n+++ b/ibis/backends/base/sql/alchemy/client.py\n@@ -1,6 +1,6 @@\n import contextlib\n import functools\n-from typing import List, Optional\n+from typing import List, Optional, Union\n \n import pandas as pd\n import sqlalchemy as sa\n@@ -9,6 +9,7 @@\n import ibis\n import ibis.expr.datatypes as dt\n import ibis.expr.schema as sch\n+import ibis.expr.types as ir\n import ibis.util as util\n from ibis.backends.base.sql.compiler import Dialect\n from ibis.client import Query, SQLClient\n@@ -229,7 +230,7 @@\n         Parameters\n         ----------\n         table_name : string\n-        data: pandas.DataFrame\n+        data : pandas.DataFrame\n         database : string, optional\n         if_exists : string, optional, default 'fail'\n             The values available are: {\u2018fail\u2019, \u2018replace\u2019, \u2018append\u2019}\n@@ -329,3 +330,93 @@\n     def version(self):\n         vstring = '.'.join(map(str, self.con.dialect.server_version_info))\n         return parse_version(vstring)\n+\n+    def insert(\n+        self,\n+        table_name: str,\n+        obj: Union[pd.DataFrame, ir.TableExpr],\n+        database: Optional[str] = None,\n+        overwrite: Optional[bool] = False,\n+    ) -> None:\n+        \"\"\"\n+        Insert the given data to a table in backend.\n+\n+        Parameters\n+        ----------\n+        table_name : string\n+            name of the table to which data needs to be inserted\n+        obj : pandas DataFrame or ibis TableExpr\n+            obj is either the dataframe (pd.DataFrame) containing data\n+            which needs to be inserted to table_name or\n+            the TableExpr type which ibis provides with data which needs\n+            to be inserted to table_name\n+        database : string, optional\n+            name of the attached database that the table is located in.\n+        overwrite : boolean, default False\n+            If True, will replace existing contents of table else not\n+\n+        Raises\n+        -------\n+        NotImplementedError\n+            Inserting data to a table from a different database is not\n+            yet implemented\n+\n+        ValueError\n+            No operation is being performed. Either the obj parameter\n+            is not a pandas DataFrame or is not a ibis TableExpr.\n+            The given obj is of type type(obj).__name__ .\n+\n+        \"\"\"\n+\n+        if database == self.database_name:\n+            # avoid fully qualified name\n+            database = None\n+\n+        if database is not None:\n+            raise NotImplementedError(\n+                'Inserting data to a table from a different database is not '\n+                'yet implemented'\n+            )\n+\n+        params = {}\n+        if self.has_attachment:\n+            # for database with attachment\n+            # see: https://github.com/ibis-project/ibis/issues/1930\n+            params['schema'] = self.database_name\n+\n+        if isinstance(obj, pd.DataFrame):\n+            obj.to_sql(\n+                table_name,\n+                self.con,\n+                index=False,\n+                if_exists='replace' if overwrite else 'append',\n+                **params,\n+            )\n+        elif isinstance(obj, ir.TableExpr):\n+            to_table_expr = self.table(table_name)\n+            to_table_schema = to_table_expr.schema()\n+\n+            if overwrite:\n+                self.drop_table(table_name, database=database)\n+                self.create_table(\n+                    table_name, schema=to_table_schema, database=database,\n+                )\n+\n+            to_table = self._get_sqla_table(table_name, schema=database)\n+\n+            from_table_expr = obj\n+\n+            with self.begin() as bind:\n+                if from_table_expr is not None:\n+                    bind.execute(\n+                        to_table.insert().from_select(\n+                            list(from_table_expr.columns),\n+                            from_table_expr.compile(),\n+                        )\n+                    )\n+        else:\n+            raise ValueError(\n+                \"No operation is being performed. Either the obj parameter \"\n+                \"is not a pandas DataFrame or is not a ibis TableExpr.\"\n+                f\"The given obj is of type {type(obj).__name__} .\"\n+            )\n", "issue": "Implement .insert() for SQLite\nSome backends implement the `.insert()` method, to insert rows into tables. SQLite doesn't, and besides being useful for users, it'd be good to have it working, and add an example in the tutorial 5, so it's useful as a reference for other backends too.\n", "before_files": [{"content": "import contextlib\nimport functools\nfrom typing import List, Optional\n\nimport pandas as pd\nimport sqlalchemy as sa\nfrom pkg_resources import parse_version\n\nimport ibis\nimport ibis.expr.datatypes as dt\nimport ibis.expr.schema as sch\nimport ibis.util as util\nfrom ibis.backends.base.sql.compiler import Dialect\nfrom ibis.client import Query, SQLClient\n\nfrom .datatypes import to_sqla_type\nfrom .geospatial import geospatial_supported\nfrom .query_builder import build_ast\nfrom .translator import AlchemyExprTranslator\n\nif geospatial_supported:\n    import geoalchemy2.shape as shape\n    import geopandas\n\n\nclass _AlchemyProxy:\n    \"\"\"\n    Wraps a SQLAlchemy ResultProxy and ensures that .close() is called on\n    garbage collection\n    \"\"\"\n\n    def __init__(self, proxy):\n        self.proxy = proxy\n\n    def __del__(self):\n        self._close_cursor()\n\n    def _close_cursor(self):\n        self.proxy.close()\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, type, value, tb):\n        self._close_cursor()\n\n    def fetchall(self):\n        return self.proxy.fetchall()\n\n\ndef _invalidates_reflection_cache(f):\n    \"\"\"Invalidate the SQLAlchemy reflection cache if `f` performs an operation\n    that mutates database or table metadata such as ``CREATE TABLE``,\n    ``DROP TABLE``, etc.\n\n    Parameters\n    ----------\n    f : callable\n        A method on :class:`ibis.sql.alchemy.AlchemyClient`\n    \"\"\"\n\n    @functools.wraps(f)\n    def wrapped(self, *args, **kwargs):\n        result = f(self, *args, **kwargs)\n\n        # only invalidate the cache after we've succesfully called the wrapped\n        # function\n        self._reflection_cache_is_dirty = True\n        return result\n\n    return wrapped\n\n\ndef _maybe_to_geodataframe(df, schema):\n    \"\"\"\n    If the required libraries for geospatial support are installed, and if a\n    geospatial column is present in the dataframe, convert it to a\n    GeoDataFrame.\n    \"\"\"\n\n    def to_shapely(row, name):\n        return shape.to_shape(row[name]) if row[name] is not None else None\n\n    if len(df) and geospatial_supported:\n        geom_col = None\n        for name, dtype in schema.items():\n            if isinstance(dtype, dt.GeoSpatial):\n                geom_col = geom_col or name\n                df[name] = df.apply(lambda x: to_shapely(x, name), axis=1)\n        if geom_col:\n            df = geopandas.GeoDataFrame(df, geometry=geom_col)\n    return df\n\n\nclass AlchemyQuery(Query):\n    def _fetch(self, cursor):\n        df = pd.DataFrame.from_records(\n            cursor.proxy.fetchall(),\n            columns=cursor.proxy.keys(),\n            coerce_float=True,\n        )\n        schema = self.schema()\n        return _maybe_to_geodataframe(schema.apply_to(df), schema)\n\n\nclass AlchemyDialect(Dialect):\n\n    translator = AlchemyExprTranslator\n\n\nclass AlchemyClient(SQLClient):\n\n    dialect = AlchemyDialect\n    query_class = AlchemyQuery\n    has_attachment = False\n\n    def __init__(self, con: sa.engine.Engine) -> None:\n        super().__init__()\n        self.con = con\n        self.meta = sa.MetaData(bind=con)\n        self._inspector = sa.inspect(con)\n        self._reflection_cache_is_dirty = False\n        self._schemas = {}\n\n    @property\n    def inspector(self):\n        if self._reflection_cache_is_dirty:\n            self._inspector.info_cache.clear()\n        return self._inspector\n\n    @contextlib.contextmanager\n    def begin(self):\n        with self.con.begin() as bind:\n            yield bind\n\n    @_invalidates_reflection_cache\n    def create_table(self, name, expr=None, schema=None, database=None):\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Creating tables from a different database is not yet '\n                'implemented'\n            )\n\n        if expr is None and schema is None:\n            raise ValueError('You must pass either an expression or a schema')\n\n        if expr is not None and schema is not None:\n            if not expr.schema().equals(ibis.schema(schema)):\n                raise TypeError(\n                    'Expression schema is not equal to passed schema. '\n                    'Try passing the expression without the schema'\n                )\n        if schema is None:\n            schema = expr.schema()\n\n        self._schemas[self._fully_qualified_name(name, database)] = schema\n        t = self._table_from_schema(\n            name, schema, database=database or self.current_database\n        )\n\n        with self.begin() as bind:\n            t.create(bind=bind)\n            if expr is not None:\n                bind.execute(\n                    t.insert().from_select(list(expr.columns), expr.compile())\n                )\n\n    def _columns_from_schema(\n        self, name: str, schema: sch.Schema\n    ) -> List[sa.Column]:\n        return [\n            sa.Column(colname, to_sqla_type(dtype), nullable=dtype.nullable)\n            for colname, dtype in zip(schema.names, schema.types)\n        ]\n\n    def _table_from_schema(\n        self, name: str, schema: sch.Schema, database: Optional[str] = None\n    ) -> sa.Table:\n        columns = self._columns_from_schema(name, schema)\n        return sa.Table(name, self.meta, *columns)\n\n    @_invalidates_reflection_cache\n    def drop_table(\n        self,\n        table_name: str,\n        database: Optional[str] = None,\n        force: bool = False,\n    ) -> None:\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Dropping tables from a different database is not yet '\n                'implemented'\n            )\n\n        t = self._get_sqla_table(table_name, schema=database, autoload=False)\n        t.drop(checkfirst=force)\n\n        assert (\n            not t.exists()\n        ), 'Something went wrong during DROP of table {!r}'.format(t.name)\n\n        self.meta.remove(t)\n\n        qualified_name = self._fully_qualified_name(table_name, database)\n\n        try:\n            del self._schemas[qualified_name]\n        except KeyError:  # schemas won't be cached if created with raw_sql\n            pass\n\n    def load_data(\n        self,\n        table_name: str,\n        data: pd.DataFrame,\n        database: str = None,\n        if_exists: str = 'fail',\n    ):\n        \"\"\"\n        Load data from a dataframe to the backend.\n\n        Parameters\n        ----------\n        table_name : string\n        data: pandas.DataFrame\n        database : string, optional\n        if_exists : string, optional, default 'fail'\n            The values available are: {\u2018fail\u2019, \u2018replace\u2019, \u2018append\u2019}\n\n        Raises\n        ------\n        NotImplementedError\n            Loading data to a table from a different database is not\n            yet implemented\n        \"\"\"\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Loading data to a table from a different database is not '\n                'yet implemented'\n            )\n\n        params = {}\n        if self.has_attachment:\n            # for database with attachment\n            # see: https://github.com/ibis-project/ibis/issues/1930\n            params['schema'] = self.database_name\n\n        data.to_sql(\n            table_name,\n            con=self.con,\n            index=False,\n            if_exists=if_exists,\n            **params,\n        )\n\n    def truncate_table(\n        self, table_name: str, database: Optional[str] = None\n    ) -> None:\n        t = self._get_sqla_table(table_name, schema=database)\n        t.delete().execute()\n\n    def list_tables(\n        self,\n        like: Optional[str] = None,\n        database: Optional[str] = None,\n        schema: Optional[str] = None,\n    ) -> List[str]:\n        \"\"\"List tables/views in the current or indicated database.\n\n        Parameters\n        ----------\n        like\n            Checks for this string contained in name\n        database\n            If not passed, uses the current database\n        schema\n            The schema namespace that tables should be listed from\n\n        Returns\n        -------\n        List[str]\n\n        \"\"\"\n        inspector = self.inspector\n        # inspector returns a mutable version of its names, so make a copy.\n        names = inspector.get_table_names(schema=schema).copy()\n        names.extend(inspector.get_view_names(schema=schema))\n        if like is not None:\n            names = [x for x in names if like in x]\n        return sorted(names)\n\n    def _execute(self, query: str, results: bool = True):\n        return _AlchemyProxy(self.con.execute(query))\n\n    @_invalidates_reflection_cache\n    def raw_sql(self, query: str, results: bool = False):\n        return super().raw_sql(query, results=results)\n\n    def _build_ast(self, expr, context):\n        return build_ast(expr, context)\n\n    def _log(self, sql):\n        try:\n            query_str = str(sql)\n        except sa.exc.UnsupportedCompilationError:\n            pass\n        else:\n            util.log(query_str)\n\n    def _get_sqla_table(self, name, schema=None, autoload=True):\n        return sa.Table(name, self.meta, schema=schema, autoload=autoload)\n\n    def _sqla_table_to_expr(self, table):\n        node = self.table_class(table, self)\n        return self.table_expr_class(node)\n\n    @property\n    def version(self):\n        vstring = '.'.join(map(str, self.con.dialect.server_version_info))\n        return parse_version(vstring)\n", "path": "ibis/backends/base/sql/alchemy/client.py"}], "after_files": [{"content": "import contextlib\nimport functools\nfrom typing import List, Optional, Union\n\nimport pandas as pd\nimport sqlalchemy as sa\nfrom pkg_resources import parse_version\n\nimport ibis\nimport ibis.expr.datatypes as dt\nimport ibis.expr.schema as sch\nimport ibis.expr.types as ir\nimport ibis.util as util\nfrom ibis.backends.base.sql.compiler import Dialect\nfrom ibis.client import Query, SQLClient\n\nfrom .datatypes import to_sqla_type\nfrom .geospatial import geospatial_supported\nfrom .query_builder import build_ast\nfrom .translator import AlchemyExprTranslator\n\nif geospatial_supported:\n    import geoalchemy2.shape as shape\n    import geopandas\n\n\nclass _AlchemyProxy:\n    \"\"\"\n    Wraps a SQLAlchemy ResultProxy and ensures that .close() is called on\n    garbage collection\n    \"\"\"\n\n    def __init__(self, proxy):\n        self.proxy = proxy\n\n    def __del__(self):\n        self._close_cursor()\n\n    def _close_cursor(self):\n        self.proxy.close()\n\n    def __enter__(self):\n        return self\n\n    def __exit__(self, type, value, tb):\n        self._close_cursor()\n\n    def fetchall(self):\n        return self.proxy.fetchall()\n\n\ndef _invalidates_reflection_cache(f):\n    \"\"\"Invalidate the SQLAlchemy reflection cache if `f` performs an operation\n    that mutates database or table metadata such as ``CREATE TABLE``,\n    ``DROP TABLE``, etc.\n\n    Parameters\n    ----------\n    f : callable\n        A method on :class:`ibis.sql.alchemy.AlchemyClient`\n    \"\"\"\n\n    @functools.wraps(f)\n    def wrapped(self, *args, **kwargs):\n        result = f(self, *args, **kwargs)\n\n        # only invalidate the cache after we've succesfully called the wrapped\n        # function\n        self._reflection_cache_is_dirty = True\n        return result\n\n    return wrapped\n\n\ndef _maybe_to_geodataframe(df, schema):\n    \"\"\"\n    If the required libraries for geospatial support are installed, and if a\n    geospatial column is present in the dataframe, convert it to a\n    GeoDataFrame.\n    \"\"\"\n\n    def to_shapely(row, name):\n        return shape.to_shape(row[name]) if row[name] is not None else None\n\n    if len(df) and geospatial_supported:\n        geom_col = None\n        for name, dtype in schema.items():\n            if isinstance(dtype, dt.GeoSpatial):\n                geom_col = geom_col or name\n                df[name] = df.apply(lambda x: to_shapely(x, name), axis=1)\n        if geom_col:\n            df = geopandas.GeoDataFrame(df, geometry=geom_col)\n    return df\n\n\nclass AlchemyQuery(Query):\n    def _fetch(self, cursor):\n        df = pd.DataFrame.from_records(\n            cursor.proxy.fetchall(),\n            columns=cursor.proxy.keys(),\n            coerce_float=True,\n        )\n        schema = self.schema()\n        return _maybe_to_geodataframe(schema.apply_to(df), schema)\n\n\nclass AlchemyDialect(Dialect):\n\n    translator = AlchemyExprTranslator\n\n\nclass AlchemyClient(SQLClient):\n\n    dialect = AlchemyDialect\n    query_class = AlchemyQuery\n    has_attachment = False\n\n    def __init__(self, con: sa.engine.Engine) -> None:\n        super().__init__()\n        self.con = con\n        self.meta = sa.MetaData(bind=con)\n        self._inspector = sa.inspect(con)\n        self._reflection_cache_is_dirty = False\n        self._schemas = {}\n\n    @property\n    def inspector(self):\n        if self._reflection_cache_is_dirty:\n            self._inspector.info_cache.clear()\n        return self._inspector\n\n    @contextlib.contextmanager\n    def begin(self):\n        with self.con.begin() as bind:\n            yield bind\n\n    @_invalidates_reflection_cache\n    def create_table(self, name, expr=None, schema=None, database=None):\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Creating tables from a different database is not yet '\n                'implemented'\n            )\n\n        if expr is None and schema is None:\n            raise ValueError('You must pass either an expression or a schema')\n\n        if expr is not None and schema is not None:\n            if not expr.schema().equals(ibis.schema(schema)):\n                raise TypeError(\n                    'Expression schema is not equal to passed schema. '\n                    'Try passing the expression without the schema'\n                )\n        if schema is None:\n            schema = expr.schema()\n\n        self._schemas[self._fully_qualified_name(name, database)] = schema\n        t = self._table_from_schema(\n            name, schema, database=database or self.current_database\n        )\n\n        with self.begin() as bind:\n            t.create(bind=bind)\n            if expr is not None:\n                bind.execute(\n                    t.insert().from_select(list(expr.columns), expr.compile())\n                )\n\n    def _columns_from_schema(\n        self, name: str, schema: sch.Schema\n    ) -> List[sa.Column]:\n        return [\n            sa.Column(colname, to_sqla_type(dtype), nullable=dtype.nullable)\n            for colname, dtype in zip(schema.names, schema.types)\n        ]\n\n    def _table_from_schema(\n        self, name: str, schema: sch.Schema, database: Optional[str] = None\n    ) -> sa.Table:\n        columns = self._columns_from_schema(name, schema)\n        return sa.Table(name, self.meta, *columns)\n\n    @_invalidates_reflection_cache\n    def drop_table(\n        self,\n        table_name: str,\n        database: Optional[str] = None,\n        force: bool = False,\n    ) -> None:\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Dropping tables from a different database is not yet '\n                'implemented'\n            )\n\n        t = self._get_sqla_table(table_name, schema=database, autoload=False)\n        t.drop(checkfirst=force)\n\n        assert (\n            not t.exists()\n        ), 'Something went wrong during DROP of table {!r}'.format(t.name)\n\n        self.meta.remove(t)\n\n        qualified_name = self._fully_qualified_name(table_name, database)\n\n        try:\n            del self._schemas[qualified_name]\n        except KeyError:  # schemas won't be cached if created with raw_sql\n            pass\n\n    def load_data(\n        self,\n        table_name: str,\n        data: pd.DataFrame,\n        database: str = None,\n        if_exists: str = 'fail',\n    ):\n        \"\"\"\n        Load data from a dataframe to the backend.\n\n        Parameters\n        ----------\n        table_name : string\n        data : pandas.DataFrame\n        database : string, optional\n        if_exists : string, optional, default 'fail'\n            The values available are: {\u2018fail\u2019, \u2018replace\u2019, \u2018append\u2019}\n\n        Raises\n        ------\n        NotImplementedError\n            Loading data to a table from a different database is not\n            yet implemented\n        \"\"\"\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Loading data to a table from a different database is not '\n                'yet implemented'\n            )\n\n        params = {}\n        if self.has_attachment:\n            # for database with attachment\n            # see: https://github.com/ibis-project/ibis/issues/1930\n            params['schema'] = self.database_name\n\n        data.to_sql(\n            table_name,\n            con=self.con,\n            index=False,\n            if_exists=if_exists,\n            **params,\n        )\n\n    def truncate_table(\n        self, table_name: str, database: Optional[str] = None\n    ) -> None:\n        t = self._get_sqla_table(table_name, schema=database)\n        t.delete().execute()\n\n    def list_tables(\n        self,\n        like: Optional[str] = None,\n        database: Optional[str] = None,\n        schema: Optional[str] = None,\n    ) -> List[str]:\n        \"\"\"List tables/views in the current or indicated database.\n\n        Parameters\n        ----------\n        like\n            Checks for this string contained in name\n        database\n            If not passed, uses the current database\n        schema\n            The schema namespace that tables should be listed from\n\n        Returns\n        -------\n        List[str]\n\n        \"\"\"\n        inspector = self.inspector\n        # inspector returns a mutable version of its names, so make a copy.\n        names = inspector.get_table_names(schema=schema).copy()\n        names.extend(inspector.get_view_names(schema=schema))\n        if like is not None:\n            names = [x for x in names if like in x]\n        return sorted(names)\n\n    def _execute(self, query: str, results: bool = True):\n        return _AlchemyProxy(self.con.execute(query))\n\n    @_invalidates_reflection_cache\n    def raw_sql(self, query: str, results: bool = False):\n        return super().raw_sql(query, results=results)\n\n    def _build_ast(self, expr, context):\n        return build_ast(expr, context)\n\n    def _log(self, sql):\n        try:\n            query_str = str(sql)\n        except sa.exc.UnsupportedCompilationError:\n            pass\n        else:\n            util.log(query_str)\n\n    def _get_sqla_table(self, name, schema=None, autoload=True):\n        return sa.Table(name, self.meta, schema=schema, autoload=autoload)\n\n    def _sqla_table_to_expr(self, table):\n        node = self.table_class(table, self)\n        return self.table_expr_class(node)\n\n    @property\n    def version(self):\n        vstring = '.'.join(map(str, self.con.dialect.server_version_info))\n        return parse_version(vstring)\n\n    def insert(\n        self,\n        table_name: str,\n        obj: Union[pd.DataFrame, ir.TableExpr],\n        database: Optional[str] = None,\n        overwrite: Optional[bool] = False,\n    ) -> None:\n        \"\"\"\n        Insert the given data to a table in backend.\n\n        Parameters\n        ----------\n        table_name : string\n            name of the table to which data needs to be inserted\n        obj : pandas DataFrame or ibis TableExpr\n            obj is either the dataframe (pd.DataFrame) containing data\n            which needs to be inserted to table_name or\n            the TableExpr type which ibis provides with data which needs\n            to be inserted to table_name\n        database : string, optional\n            name of the attached database that the table is located in.\n        overwrite : boolean, default False\n            If True, will replace existing contents of table else not\n\n        Raises\n        -------\n        NotImplementedError\n            Inserting data to a table from a different database is not\n            yet implemented\n\n        ValueError\n            No operation is being performed. Either the obj parameter\n            is not a pandas DataFrame or is not a ibis TableExpr.\n            The given obj is of type type(obj).__name__ .\n\n        \"\"\"\n\n        if database == self.database_name:\n            # avoid fully qualified name\n            database = None\n\n        if database is not None:\n            raise NotImplementedError(\n                'Inserting data to a table from a different database is not '\n                'yet implemented'\n            )\n\n        params = {}\n        if self.has_attachment:\n            # for database with attachment\n            # see: https://github.com/ibis-project/ibis/issues/1930\n            params['schema'] = self.database_name\n\n        if isinstance(obj, pd.DataFrame):\n            obj.to_sql(\n                table_name,\n                self.con,\n                index=False,\n                if_exists='replace' if overwrite else 'append',\n                **params,\n            )\n        elif isinstance(obj, ir.TableExpr):\n            to_table_expr = self.table(table_name)\n            to_table_schema = to_table_expr.schema()\n\n            if overwrite:\n                self.drop_table(table_name, database=database)\n                self.create_table(\n                    table_name, schema=to_table_schema, database=database,\n                )\n\n            to_table = self._get_sqla_table(table_name, schema=database)\n\n            from_table_expr = obj\n\n            with self.begin() as bind:\n                if from_table_expr is not None:\n                    bind.execute(\n                        to_table.insert().from_select(\n                            list(from_table_expr.columns),\n                            from_table_expr.compile(),\n                        )\n                    )\n        else:\n            raise ValueError(\n                \"No operation is being performed. Either the obj parameter \"\n                \"is not a pandas DataFrame or is not a ibis TableExpr.\"\n                f\"The given obj is of type {type(obj).__name__} .\"\n            )\n", "path": "ibis/backends/base/sql/alchemy/client.py"}]}