Catalog - Fenic

The Catalog API provides methods to manage database objects in Fenic, including catalogs, databases, tables, views, and user-defined tools. It also provides read-only access to system tables such as query metrics.

Access

Access the catalog through a session’s catalog property:

session = fc.Session.get_or_create()
session.catalog.list_tables()

Catalog Management

create_catalog()

Creates a new catalog.

catalog_name

str

required

Name of the catalog to create.

ignore_if_exists

bool

default:"True"

If True, return False when the catalog already exists. If False, raise an error.

Returns: bool - True if created successfully, False if already exists and ignore_if_exists=True

session.catalog.create_catalog('my_catalog')  # Returns: True

set_current_catalog()

Sets the current catalog.

catalog_name

str

required

Name of the catalog to set as current.

session.catalog.set_current_catalog('my_catalog')

get_current_catalog()

Returns the name of the current catalog. Returns: str - The current catalog name

catalog = session.catalog.get_current_catalog()  # Returns: 'default'

list_catalogs()

Returns a list of available catalogs. Returns: List[str] - List of catalog names

catalogs = session.catalog.list_catalogs()
# Returns: ['default', 'my_catalog', 'other_catalog']

does_catalog_exist()

Checks if a catalog exists.

catalog_name

str

required

Name of the catalog to check.

Returns: bool - True if the catalog exists

exists = session.catalog.does_catalog_exist('my_catalog')

drop_catalog()

Drops a catalog.

catalog_name

str

required

Name of the catalog to drop.

ignore_if_not_exists

bool

default:"True"

If True, silently return if the catalog doesn’t exist. If False, raise an error.

Returns: bool - True if dropped successfully, False if didn’t exist and ignore_if_not_exists=True

session.catalog.drop_catalog('my_catalog')

Database Management

create_database()

Creates a new database.

database_name

str

required

Fully qualified or relative database name to create.

ignore_if_exists

bool

default:"True"

If True, return False when the database already exists. If False, raise an error.

Returns: bool - True if created successfully, False if already exists and ignore_if_exists=True

session.catalog.create_database('my_database')  # Returns: True

set_current_database()

Sets the current database.

database_name

str

required

Fully qualified or relative database name to set as current.

session.catalog.set_current_database('my_database')

get_current_database()

Returns the name of the current database. Returns: str - The current database name

db = session.catalog.get_current_database()  # Returns: 'default'

list_databases()

Returns a list of databases in the current catalog. Returns: List[str] - List of database names

databases = session.catalog.list_databases()
# Returns: ['default', 'my_database', 'other_database']

does_database_exist()

Checks if a database exists.

database_name

str

required

Fully qualified or relative database name to check.

Returns: bool - True if the database exists

exists = session.catalog.does_database_exist('my_database')

drop_database()

Drops a database.

database_name

str

required

Fully qualified or relative database name to drop.

cascade

bool

default:"False"

If True, drop all tables in the database.

ignore_if_not_exists

bool

default:"True"

If True, silently return if the database doesn’t exist. If False, raise an error.

Returns: bool - True if dropped successfully, False if didn’t exist and ignore_if_not_exists=True

session.catalog.drop_database('my_database', cascade=True)

Table Management

create_table()

Creates a new table.

table_name

str

required

Fully qualified or relative table name to create.

schema

Schema

required

Schema of the table to create.

ignore_if_exists

bool

default:"True"

If True, return False when the table already exists. If False, raise an error.

description

str

default:"None"

Description of the table.

Returns: bool - True if created successfully, False if already exists and ignore_if_exists=True

from fenic import Schema, ColumnField, IntegerType, StringType

session.catalog.create_table(
    'my_table', 
    Schema([
        ColumnField('id', IntegerType),
        ColumnField('name', StringType)
    ]),
    description='My table description'
)  # Returns: True

list_tables()

Returns a list of tables in the current database. Returns: List[str] - List of table names

tables = session.catalog.list_tables()
# Returns: ['table1', 'table2', 'table3']

does_table_exist()

Checks if a table exists.

table_name

str

required

Fully qualified or relative table name to check.

Returns: bool - True if the table exists

exists = session.catalog.does_table_exist('my_table')

describe_table()

Returns the schema and description of a table.

table_name

str

required

Fully qualified or relative table name to describe.

Returns: DatasetMetadata - Object containing schema and description

metadata = session.catalog.describe_table('my_table')
print(metadata.schema)
print(metadata.description)

set_table_description()

Set or unset the description for a table.

table_name

str

required

Fully qualified or relative table name.

description

str

default:"None"

The description to set for the table.

session.catalog.set_table_description('my_table', 'Updated description')

drop_table()

Drops a table.

table_name

str

required

Fully qualified or relative table name to drop.

ignore_if_not_exists

bool

default:"True"

If True, return False when the table doesn’t exist. If False, raise an error.

Returns: bool - True if dropped successfully, False if didn’t exist and ignore_if_not_exists=True

session.catalog.drop_table('my_table')

View Management

list_views()

Returns a list of views in the current database. Returns: List[str] - List of view names

views = session.catalog.list_views()
# Returns: ['view1', 'view2', 'view3']

does_view_exist()

Checks if a view exists.

view_name

str

required

Fully qualified or relative view name to check.

Returns: bool - True if the view exists

exists = session.catalog.does_view_exist('my_view')

describe_view()

Returns the schema and description of a view.

view_name

str

required

Fully qualified or relative view name to describe.

Returns: DatasetMetadata - Object containing schema and description

metadata = session.catalog.describe_view('my_view')
print(metadata.schema)
print(metadata.description)

set_view_description()

Set the description for a view.

view_name

str

required

Fully qualified or relative view name.

description

str

default:"None"

The description to set for the view.

session.catalog.set_view_description('my_view', 'Customer summary view')

drop_view()

Drops a view.

view_name

str

required

Fully qualified or relative view name to drop.

ignore_if_not_exists

bool

default:"True"

If True, return False when the view doesn’t exist. If False, raise an error.

Returns: bool - True if dropped successfully, False if didn’t exist and ignore_if_not_exists=True

session.catalog.drop_view('my_view')

Tool Management

create_tool()

Creates a new user-defined tool in the current catalog.

tool_name

str

required

The name of the tool.

tool_description

str

required

The description of the tool.

tool_query

DataFrame

required

The query to execute when the tool is called.

tool_params

List[ToolParam]

required

The parameters of the tool.

result_limit

int

default:"50"

The maximum number of rows to return from the tool.

ignore_if_exists

bool

default:"True"

If True, return False when the tool already exists. If False, raise an error.

Returns: bool - True if created successfully, False if already exists and ignore_if_exists=True

from fenic.core.mcp.types import ToolParam

df = session.create_dataframe(...)

session.catalog.create_tool(
    tool_name="my_tool",
    tool_description="A tool that does something",
    tool_query=df,
    result_limit=100,
    tool_params=[ToolParam(
        name="param1", 
        description="A parameter", 
        allowed_values=["value1", "value2"]
    )],
)

list_tools()

Lists the tools available in the current catalog. Returns: List[UserDefinedTool] - List of tools

tools = session.catalog.list_tools()

describe_tool()

Returns the tool with the specified name.

tool_name

str

required

The name of the tool to get.

Returns: UserDefinedTool - The tool object

tool = session.catalog.describe_tool('my_tool')

drop_tool()

Drops a tool from the current catalog.

tool_name

str

required

The name of the tool to drop.

ignore_if_not_exists

bool

default:"True"

If True, return False when the tool doesn’t exist. If False, raise an error.

Returns: bool - True if dropped successfully, False if didn’t exist and ignore_if_not_exists=True

session.catalog.drop_tool('my_tool')

System Tables

Query Metrics Table (Local Sessions Only)

Query metrics are recorded for each session and stored locally in fenic_system.query_metrics. You can load this table into a DataFrame for analysis.

# Load all metrics for the current application
metrics_df = session.table("fenic_system.query_metrics")

# Show the 10 most recent queries
recent_queries = session.sql("""
    SELECT *
    FROM {df}
    ORDER BY CAST(end_ts AS TIMESTAMP) DESC
    LIMIT 10
""", df=metrics_df)
recent_queries.show()

# Find query metrics for a specific session with non-zero LM costs
specific_session_queries = session.sql("""
    SELECT *
    FROM {df}
    WHERE session_id = '9e7e256f-fad9-4cd9-844e-399d795aaea0'
        AND total_lm_cost > 0
    ORDER BY CAST(end_ts AS TIMESTAMP) ASC
""", df=metrics_df)
specific_session_queries.show()

# Aggregate total LM costs and requests in a time window
metrics_window = session.sql("""
    SELECT
        CAST(SUM(total_lm_cost) AS DOUBLE) AS total_lm_cost_in_window,
        CAST(SUM(total_lm_requests) AS DOUBLE) AS total_lm_requests_in_window
    FROM {df}
    WHERE CAST(end_ts AS TIMESTAMP) BETWEEN
        CAST('2025-08-29 10:00:00' AS TIMESTAMP)
        AND CAST('2025-08-29 12:00:00' AS TIMESTAMP)
""", df=metrics_df)
metrics_window.show()

Core

Functions

I/O

Types

Configuration

MCP

​Access

​Catalog Management

​create_catalog()

​set_current_catalog()

​get_current_catalog()

​list_catalogs()

​does_catalog_exist()

​drop_catalog()

​Database Management

​create_database()

​set_current_database()

​get_current_database()

​list_databases()

​does_database_exist()

​drop_database()

​Table Management

​create_table()

​list_tables()

​does_table_exist()

​describe_table()

​set_table_description()

​drop_table()

​View Management

​list_views()

​does_view_exist()

​describe_view()

​set_view_description()

​drop_view()

​Tool Management

​create_tool()

​list_tools()

​describe_tool()

​drop_tool()

​System Tables

​Query Metrics Table (Local Sessions Only)

​See Also

Build docs developers (and LLMs) love

Access

Catalog Management

create_catalog()

set_current_catalog()

get_current_catalog()

list_catalogs()

does_catalog_exist()

drop_catalog()

Database Management

create_database()

set_current_database()

get_current_database()

list_databases()

does_database_exist()

drop_database()

Table Management

create_table()

list_tables()

does_table_exist()

describe_table()

set_table_description()

drop_table()

View Management

list_views()

does_view_exist()

describe_view()

set_view_description()

drop_view()

Tool Management

create_tool()

list_tools()

describe_tool()

drop_tool()

System Tables

Query Metrics Table (Local Sessions Only)

See Also