Functions

This section describes the main API functions provided by morphdb-utils, allowing you to perform database operations and file manipulations in a simple and intuitive manner.

The APIs of morphdb-utils provide the following functions:

• Execute SQL queries with the execute_sql function.
• Retrieve information from other cells with the ref function.
• Read the contents of a directory with the read_dir function.
• Obtain file paths or URLs with the get_file function.
• Obtain data paths or URLs with the get_data_path function.
• Create tables and insert data with the create_table function.
• Insert records into a table with the insert_records function.
• Update table records with the update_records function.
• Generate reports with the generate_report function.
• Send emails with the send_email function.
• Load data with the load_data function.

Types of APIs

execute_sql

Executes the specified SQL query and returns the result as a DataFrame.

from typing import Dict
import pandas as pd
from morphdb_utils.api import execute_sql

data: Dict[str, pd.DataFrame] = {
    "now": execute_sql("SELECT NOW();")
}

def main(data: Dict[str, pd.DataFrame]) -> None:
    print(data["now"])
    return None

ref

Retrieves information from the specified reference cell in the form of a RefResponse object.

from typing import Dict
import pandas as pd
from morphdb_utils.api import ref
from morphdb_utils.type import RefResponse

data: Dict[str, RefResponse] = {
    "example_python_cell": ref("example_python_cell"),
}

def main(data: Dict[str, RefResponse]) -> RefResponse:
    return data["example_python_cell"]

read_dir

Returns a list of objects within the specified directory.

from typing import Dict
import pandas as pd
from morphdb_utils.api import read_dir

data: Dict[str, pd.DataFrame] = {
    "src_dir": read_dir("./src")
}

def main(data: Dict[str, pd.DataFrame]) -> None:
		print(data["src_dir"])
		return None

get_file

Obtains the absolute path or URL of the specified file.

from typing import Dict
import pandas as pd
from morphdb_utils.api import get_file

data: Dict[str, pd.DataFrame] = {
    "file": get_file("./src/example_python_cell.py")
}

def main(data: Dict[str, pd.DataFrame]) -> None:
		print(data["file"])
		return None

get_data_path

Obtains the absolute path or URL of the output data for the specified file.

from typing import Dict
import pandas as pd
from morphdb_utils.api import ref, get_data_path

data: Dict[str, pd.DataFrame] = {
    "data_path": get_data_path(ref("example_python_cell"))
}

def main(data: Dict[str, pd.DataFrame]) -> None:
		print(data["data_path"])
		return None

create_table

Creates a new table from the specified DataFrame and inserts the data.

from typing import Dict
import pandas as pd
from morphdb_utils.annotations import transform
from morphdb_utils.api import create_table

data: Dict[str, pd.DataFrame] = {}

@transform
def main(data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
    example_data = {
        "Name": ["John Doe", "Jane Smith", "Emily Zhang"],
        "Age": [28, 34, 22],
    }
    df = pd.DataFrame(example_data)
    create_table(df, table_name="users")
    return df

insert_records

Inserts data from the specified DataFrame into an existing table.

from typing import Dict
import pandas as pd
from morphdb_utils.annotations import transform
from morphdb_utils.api import insert_records

data: Dict[str, pd.DataFrame] = {}

@transform
def main(data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
    example_data = {
	    "Name": ["Alice Brown", "Bob Davis"],
	    "Age": [30, 40],
    }
    df = pd.DataFrame(example_data)
    insert_records(df, table_name="users")
    return df

update_records

Updates data in an existing table with the specified DataFrame.

from typing import Dict
import pandas as pd
from morphdb_utils.annotations import transform
from morphdb_utils.api import update_records

data: Dict[str, pd.DataFrame] = {}

@transform
def main(data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
    example_data = {
	    "Name": ["Alice Brown", "Bob Davis"],
	    "Age": [31, 41],
    }
    df = pd.DataFrame(example_data)
    update_records(df, table_name="users", key_columns=["Name"])
    return df

generate_report

Generates a report from the specified reference cell.

from typing import Dict
import pandas as pd
from morphdb_utils.api import generate_report, ref
from morphdb_utils.type import RefResponse

data: Dict[str, RefResponse] = {
    "example_python_cell": ref("example_python_cell"),
}

def main(data: Dict[str, RefResponse]) -> None:
    refs = [data["example_python_cell"]]
    generate_report(refs, prompt="Generate a summary report")
    return None

send_email

Sends an email with the specified reference cell as an attachment.

from typing import Dict
import pandas as pd
from morphdb_utils.api import send_email, ref
from morphdb_utils.type import RefResponse

data: Dict[str, RefResponse] = {
    "example_python_cell": ref("example_python_cell"),
}

def main(data: Dict[str, RefResponse]) -> None:
    refs = [data["example_python_cell"]]
    emails = ["morph@queue-inc.com"]
    subject = "User Data Report"
    body = "Please find the attached user data report."
    send_email(refs, emails, subject, body)
    return None

load_data

Loads data from the specified file path or reference cell.

from typing import Dict
import pandas as pd
from morphdb_utils.api import load_data, ref
from morphdb_utils.type import RefResponse

data: Dict[str, pd.DataFrame] = {
    "example_python_cell": load_data(ref("example_python_cell")),
}

def main(data: Dict[str, pd.DataFrame]) -> pd.DataFrame:
    return data["example_python_cell"]