# Lakehouse Manager Dataset connector

> Using Dataset connector, you may query, insert, update or delete content programmatically.

* [Connect to the Lakehouse Manager](#connect-to-the-lakehouse-manager)
* [Connector methods](#connector-methods)
    * [list()](#list)
    * [query(sql, ...)](#querysql-limit-1-return_typedataframe)
    * [select(table_name, ...)](#selecttable_name-conds-limit-1-return_typedataframe-)
    * [update(table_name, ...)](#updatetable_name-set-condsnone-ignorefalse)
    * [delete(table_name, ...)](#deletetable_name-conds)
    * [truncate(table_name)](#truncatetable_name)
    * [insert(table_name, rows, ...)](#inserttable_name-rows-odku0-returnwarningsfalse)
* [LogicalObject methods](#logicalobject-methods)
    * [build()](#buildobjects)
    * [create_from_physical()](#create_from_physicaltable-dataset39default_dataset39-kwargs)
    * [get()](#getname)
    * [list()](#listdataset_name39default_dataset39)
    * [remove()](#removename)

---

## Connect to the Lakehouse Manager

In order to create a Connector and use it to interact with a [Lakehouse Manager table](/en/product/lakehouse-manager/index), you can connect to the **Default Dataset**  or to a **Custom Dataset**. 

Below, you will see the connection strings used for each type: 

```python
from forepaas.dwh import connect

cn_default = connect("dwh/default_dataset/")
cn_custom = connect("dwh/custom_dataset_name/")
```

After that you can use the `cn_default.list()` method to see the tables available in your Lakehouse Manager and then `cn_default.select(...)` or `cn_default.query(...)` to get the data from the table you want.

See the next section of this article for additional details.

!> Note that **tables need to be loaded in the Lakehouse Manager** before using the `Connector.list()` method and other Connector object methods. In other words, you have to create a table first in the Lakehouse Manager in order to use a table in the SDK.

---
## Connector methods

### list()
Get the list of tables in the dataset.

**Output**

| Type  | Description | Example |
| :---: | :---        | :---    |
| list | list of table names | ```["chicago_calendar_full", "stations_rides"]``` |

---

### query(sql, limit=-1, return_type='dataframe')

Execute a SQL query on a compatible `source`, returns a dataframe (default), an iterable cursor, or a list of dict.

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| sql   | str| SQL query to execute | ```select * from stations_rides``` |
| limit | int | maximum number of results (-1: no limit) | -1 |
| return_type | str | type of return | 'dataframe', 'cursor' or 'dict' |

**Output**

| Type  | Description | Example |
| :---: | :---        | :---    |
| mixed | results in the choosen type (dataframe, cursor, list[dict] |      |

---

### select(table_name, conds={}, limit=-1, return_type='dataframe' )

Extract data from a table with simple filters, returns a dataframe (default), an iterable cursor, or a list of dict.

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | source table name | `stations_rides` |
| conds | dict | Dict of values to use as filters. `Keys` are attributes to filter with `values` as the filtering values. `List` produces a IN operator, otherwise produce a `=` operator | None / ```{"date":["01/11/2013", "01/12/2013"], "station_name":"Harlem-Lake"}``` |
| limit | int | maximum number of results. (-1: no limit) | -1 |
| return_type | string | type of return | 'dataframe', 'cursor' or 'dict' |


**Output**

| Type  | Description | Example |
| :---: | :---        | :---    |
| mixed | results in the choosen type (dataframe, cursor, list[dict] |      |

---

### update(table_name, set, conds=None, ignore=False)
Performs an `UPDATE` SQL query through simple parameters and return the number of affected rows.

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | name of the source table | `'stations_rides'` |
| set | dict | Dict of values to update. Key as the field name to update, values as the value to update. | ```{“rides”:0}``` |
| conds | dict | Dict of values to use as filters. `Keys` are attributes to filter with `values` as the filtering values. `List` produces a IN operator, otherwise produce a `=` operator | None / ```{"date":["01/11/2013", "01/12/2013"], "station_name":"Harlem-Lake"}``` |
| ignore | boolean | If `True`, performs an `UPDATE IGNORE` query. Otherwise `UPDATE` query. | `True`, `False` |

**Outputs**

| Type  | Description | Example |
| :---: | :---        | :---    |
| int   | Number of updated rows (if DBMS compatible) | 42 |


---

### delete(table_name, conds)
Performs a deletion based on a specific query with filtering conditions. Returns the number of deleted rows.  
If you don’t need any filter, use the `truncate` method described below.

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | name of the source table | `'stations_rides'` |
| conds | dict | Dict of values to use as filters. `Keys` are attributes to filter with `values` as the filtering values. `List` produces a IN operator, otherwise produce a `=` operator | ```{"station_name":"Harlem-Lake"}``` |

**Outputs**

| Type  | Description | Example |
| :---: | :---        | :---    |
| int   | Number of deleted rows (if DBMS compatible) | 42 |

---

### truncate(table_name)
Truncates (clears) all rows from a table.

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | name of the source table | `'stations_rides'` |

**Outputs**

| Type  | Description | Example |
| :---: | :---        | :---    |
| boolean | True if truncate succeeded. False if table does not exist. Raise Exception if anything else occurred. | `True`, `False` |

---

### insert(table_name, rows, odku=0, returnWarnings=False)

Executes an insert query, adds `on duplicate key update` (ODKU) operation optionally, and returns statistics about the insertion of the batch.  
If you activate the ODKU option, the query will automatically performs updates on existing rows instead of returning errors, regarding the existing primary key values in each row that you are inserting.

!> Please note that you **MUST** provide, for each row that you want to insert, at least the primary key of the destination table.

> We encourage you to use the module **bulk_insert** instead of this function, as documented in the Connector methods section of this page, because it has better management of data chunk and insertion.

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | name of the source table | `'stations_rides'` |
| rows | dataframe | Dataframe to insert or update. | ```pd.DataFrame([ {"station_id": 1, "station_name":"First Station"}, {"station_id":2, "name":"Second Station"}])``` |
| odku | boolean  | `False` = Don’t perform an ODKU; `True` = Performs an ODKU | `True`, `False` |
| returnWarnings | boolean | If True, return warnings aggregated and counted for each type of warning strings returned | True / False |

**Outputs**

| Type  | Description | Example |
| :---: | :---        | :---    |
| dict | Statistics about the inserted batch. | ```{"inserts":13, "skipped":10, "updates":3, "records": 30, "affected":4, "warnings": {"integer truncated":4}}``` |

---

### drop_table(table_name)

drops the table from the query engine

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | name of the source table | `'stations_rides'` |

**Outputs**

| Type  | Description | Example |
| :---: | :---        | :---    |
| boolean | True if drop succeeded. False if table does not exist. Raise Exception if anything else occurred. | `True`, `False` |

---

### get_table_schema(table_name, catalog_name, schema_name, strict=True)

Return a table schema as a list of `forepaas.dwh.attributes.AttributeSchema` object

**Input Parameters**

| Name  | Type  | Description | Example |
| :---  | :---: | :---        | :---    |
| table_name | str | name of the source table | `'stations_rides'` |
| catalog_name | str | name of the catalog. If `None`, the default connector's catalog is used. | `'default_dataset'` |
| schema_name | str | name of the schema. If `None` uniqueness of the table will be checked across all catalog's schemas. | `'default_schema'` |
| strict | bool | if true, will raise an error if a column has an unsupported type. Otherwise will just log a warning. | True / False |

**Output Parameters**

| Type  | Description |
| :---: | :---        |
| list | AttributeSchema object can have 4 attributes: `name`, `data_type`, `nullable` & `default_value` |

**Example Output**

```python
[
    AttributeSchema(name='id', data_type=DataType.INT, nullable=False),
    AttributeSchema(name='name', data_type=DataType.VARCHAR, nullable=True),
    AttributeSchema(name='email', data_type=DataType.VARCHAR, nullable=False, default_value=''),
    AttributeSchema(name='created_at', data_type=DataType.TIMESTAMP, nullable=False)
]
```

---

## LogicalObject Methods

In order to use the `LogicalObject` methods you will need to first import it using:

```python
from forepaas.dwh.logical import LogicalObject

logical = LogicalObject() 
logical.list()
```

### build(objects)

Launches a logical build in the Lakehouse Manager.

**Input Parameters**

| Name | Type | Description | Example |
| :--- | :---: | :--- | :--- |
| objects | `list` | A list of logical objects to build. | `['user_data', 'product_metrics']` |

**Outputs**

| Type | Description |
| :---: | :--- |
| `int` | The status code of the build result. |

**Output Example**

```python
200 # Build successful
```

---

### create_from_physical(table, dataset='default_dataset', **kwargs)

Creates a logical object based on an existing physical table.

**Note**: Currently works only with Trino and PostgreSQL catalogs.

**Input Parameters**

| Name | Type | Description | Example |
| :--- | :---: | :--- | :--- |
| table | `str` | The name of the physical table. | `'raw_customer_data'` |
| dataset | `str` | The dataset name. Works only with Trino/PostgreSQL datasets. Default: `'default_dataset'`. | `'my_trino_dataset'` |
| kwargs | `dict` | Named arguments passed as JSON content to the API for additional configuration. | `{'owner': 'data_team', 'tags': ['customers', 'raw']}` |

**Outputs**

| Type | Description |
| :---: | :--- |
| `LogicalObject` | The created logical object. |

---

### get(name)

Retrieves a logical object from the Lakehouse Manager.

**Input Parameters**

| Name | Type | Description | Example |
| :--- | :---: | :--- | :--- |
| name | `str` | The name of the logical object to retrieve. | `'sales_dashboard_view'` |

**Outputs**

| Type | Description |
| :---: | :--- |
| `LogicalObject` | The retrieved logical object. |

---

### list(dataset_name='default_dataset')

Lists all logical objects in the Lakehouse Manager.

**Input Parameters**

| Name | Type | Description | Example |
| :--- | :---: | :--- | :--- |
| dataset_name | `str` | The dataset to filter logical objects by. Defaults to `"default_dataset"`. | `'default_dataset'` |

**Outputs**

| Type | Description |
| :---: | :--- |
| `list[LogicalObject]` | A list of logical objects. |

---

### remove(name)

Removes a logical object from the Lakehouse Manager.

**Input Parameters**

| Name | Type | Description | Example |
| :--- | :---: | :--- | :--- |
| name | `str` | The name of the logical object to remove. | `'old_temp_table'` |

**Outputs**

| Type | Description |
| :---: | :--- |
| `object` | The status of the removal operation. |

**Output Example**

```python
{'success': True, 'info': None, '_id': '690325793a04befefec45bf1', 'uid': None}
```

---

## Complementary notes

Due to lacks of metrics collected by PostgreSQL or Snowflake libraries, the statistics returned by Data Platform connectors are calculated based on limited factors.
- When no error happens, *inserts* and *skipped* rows are calculated based on total rows before and after inserting, as well as the *records* (length of data to insert) provided. For example, if we try to insert 3 rows into a table of 5 rows, and the result is 7 rows in total, we consider 2 rows are *inserted* and 1 row *skipped*, while *records* is 3.
- If error happens in a batch of data, the whole batch will be marked as warnings.
- Note for Snowflake: statistics returned by `insert_many()` and `insert_dataframe()` wont have *warnings* calculated, all rows fail to insert or skipped will be marked as *skipped*