fenic.api.catalog

Catalog API for managing database objects in Fenic.

Classes:

• Catalog –

  Entry point for catalog operations.

Catalog

Catalog(catalog: BaseCatalog)

Entry point for catalog operations.

The Catalog provides methods to interact with and manage database tables, including listing available tables, describing table schemas, and dropping tables.

Basic usage

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

# Set the current catalog
session.catalog.set_current_catalog('my_catalog')
# Returns: None

# Create a new database
session.catalog.create_database('my_database')
# Returns: True

# Use the new database
session.catalog.set_current_database('my_database')
# Returns: None

# Create a new table
session.catalog.create_table('my_table', Schema([
    ColumnField('id', IntegerType),
]))
# Returns: True

Initialize a Catalog instance.

Parameters:

• catalog (BaseCatalog) –

  The underlying catalog implementation.

Methods:

• create_catalog –

  Creates a new catalog.

• create_database –

  Creates a new database.

• create_table –

  Creates a new table.

• describe_table –

  Returns the schema of the specified table.

• does_catalog_exist –

  Checks if a catalog with the specified name exists.

• does_database_exist –

  Checks if a database with the specified name exists.

• does_table_exist –

  Checks if a table with the specified name exists.

• drop_catalog –

  Drops a catalog.

• drop_database –

  Drops a database.

• drop_table –

  Drops the specified table.

• get_current_catalog –

  Returns the name of the current catalog.

• get_current_database –

  Returns the name of the current database in the current catalog.

• list_catalogs –

  Returns a list of available catalogs.

• list_databases –

  Returns a list of databases in the current catalog.

• list_tables –

  Returns a list of tables stored in the current database.

• set_current_catalog –

  Sets the current catalog.

• set_current_database –

  Sets the current database.

Source code in src/fenic/api/catalog.py

def __init__(self, catalog: BaseCatalog):
    """Initialize a Catalog instance.

    Args:
        catalog: The underlying catalog implementation.
    """
    self.catalog = catalog

create_catalog

create_catalog(catalog_name: str, ignore_if_exists: bool = True) -> bool

Creates a new catalog.

Parameters:

• catalog_name (str) –

  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 when the catalog already exists. Defaults to True.

Raises:

• CatalogAlreadyExistsError –

  If the catalog already exists and ignore_if_exists is False.

Returns:

• bool ( bool ) –

  True if the catalog was created successfully, False if the catalog

• bool –

  already exists and ignore_if_exists is True.

Create a new catalog

# Create a new catalog named 'my_catalog'
session.catalog.create_catalog('my_catalog')
# Returns: True

Create an existing catalog with ignore_if_exists

# Try to create an existing catalog with ignore_if_exists=True
session.catalog.create_catalog('my_catalog', ignore_if_exists=True)
# Returns: False

Create an existing catalog without ignore_if_exists

# Try to create an existing catalog with ignore_if_exists=False
session.catalog.create_catalog('my_catalog', ignore_if_exists=False)
# Raises: CatalogAlreadyExistsError

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def create_catalog(self, catalog_name: str, ignore_if_exists: bool = True) -> bool:
    """Creates a new catalog.

    Args:
        catalog_name (str): Name of the catalog to create.
        ignore_if_exists (bool): If True, return False when the catalog already exists.
            If False, raise an error when the catalog already exists.
            Defaults to True.

    Raises:
        CatalogAlreadyExistsError: If the catalog already exists and ignore_if_exists is False.

    Returns:
        bool: True if the catalog was created successfully, False if the catalog
        already exists and ignore_if_exists is True.

    Example: Create a new catalog
        ```python
        # Create a new catalog named 'my_catalog'
        session.catalog.create_catalog('my_catalog')
        # Returns: True
        ```

    Example: Create an existing catalog with ignore_if_exists
        ```python
        # Try to create an existing catalog with ignore_if_exists=True
        session.catalog.create_catalog('my_catalog', ignore_if_exists=True)
        # Returns: False
        ```

    Example: Create an existing catalog without ignore_if_exists
        ```python
        # Try to create an existing catalog with ignore_if_exists=False
        session.catalog.create_catalog('my_catalog', ignore_if_exists=False)
        # Raises: CatalogAlreadyExistsError
        ```
    """
    return self.catalog.create_catalog(catalog_name, ignore_if_exists)

create_database

create_database(database_name: str, ignore_if_exists: bool = True) -> bool

Creates a new database.

Parameters:

• database_name (str) –

  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 when the database already exists. Defaults to True.

Raises:

• DatabaseAlreadyExistsError –

  If the database already exists and ignore_if_exists is False.

Returns:

• bool ( bool ) –

  True if the database was created successfully, False if the database

• bool –

  already exists and ignore_if_exists is True.

Create a new database

# Create a new database named 'my_database'
session.catalog.create_database('my_database')
# Returns: True

Create an existing database with ignore_if_exists

# Try to create an existing database with ignore_if_exists=True
session.catalog.create_database('my_database', ignore_if_exists=True)
# Returns: False

Create an existing database without ignore_if_exists

# Try to create an existing database with ignore_if_exists=False
session.catalog.create_database('my_database', ignore_if_exists=False)
# Raises: DatabaseAlreadyExistsError

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def create_database(
    self, database_name: str, ignore_if_exists: bool = True
) -> bool:
    """Creates a new database.

    Args:
        database_name (str): Fully qualified or relative database name to create.
        ignore_if_exists (bool): If True, return False when the database already exists.
            If False, raise an error when the database already exists.
            Defaults to True.

    Raises:
        DatabaseAlreadyExistsError: If the database already exists and ignore_if_exists is False.

    Returns:
        bool: True if the database was created successfully, False if the database
        already exists and ignore_if_exists is True.

    Example: Create a new database
        ```python
        # Create a new database named 'my_database'
        session.catalog.create_database('my_database')
        # Returns: True
        ```

    Example: Create an existing database with ignore_if_exists
        ```python
        # Try to create an existing database with ignore_if_exists=True
        session.catalog.create_database('my_database', ignore_if_exists=True)
        # Returns: False
        ```

    Example: Create an existing database without ignore_if_exists
        ```python
        # Try to create an existing database with ignore_if_exists=False
        session.catalog.create_database('my_database', ignore_if_exists=False)
        # Raises: DatabaseAlreadyExistsError
        ```
    """
    return self.catalog.create_database(database_name, ignore_if_exists)

create_table

create_table(table_name: str, schema: Schema, ignore_if_exists: bool = True) -> bool

Creates a new table.

Parameters:

• table_name (str) –

  Fully qualified or relative table name to create.

• schema (Schema) –

  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 when the table already exists. Defaults to True.

Returns:

• bool ( bool ) –

  True if the table was created successfully, False if the table

• bool –

  already exists and ignore_if_exists is True.

Raises:

• TableAlreadyExistsError –

  If the table already exists and ignore_if_exists is False

Create a new table

# Create a new table with an integer column
session.catalog.create_table('my_table', Schema([
    ColumnField('id', IntegerType),
]))
# Returns: True

Create an existing table with ignore_if_exists

# Try to create an existing table with ignore_if_exists=True
session.catalog.create_table('my_table', Schema([
    ColumnField('id', IntegerType),
]), ignore_if_exists=True)
# Returns: False

Create an existing table without ignore_if_exists

# Try to create an existing table with ignore_if_exists=False
session.catalog.create_table('my_table', Schema([
    ColumnField('id', IntegerType),
]), ignore_if_exists=False)
# Raises: TableAlreadyExistsError

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def create_table(
    self, table_name: str, schema: Schema, ignore_if_exists: bool = True
) -> bool:
    """Creates a new table.

    Args:
        table_name (str): Fully qualified or relative table name to create.
        schema (Schema): Schema of the table to create.
        ignore_if_exists (bool): If True, return False when the table already exists.
            If False, raise an error when the table already exists.
            Defaults to True.

    Returns:
        bool: True if the table was created successfully, False if the table
        already exists and ignore_if_exists is True.

    Raises:
        TableAlreadyExistsError: If the table already exists and ignore_if_exists is False

    Example: Create a new table
        ```python
        # Create a new table with an integer column
        session.catalog.create_table('my_table', Schema([
            ColumnField('id', IntegerType),
        ]))
        # Returns: True
        ```

    Example: Create an existing table with ignore_if_exists
        ```python
        # Try to create an existing table with ignore_if_exists=True
        session.catalog.create_table('my_table', Schema([
            ColumnField('id', IntegerType),
        ]), ignore_if_exists=True)
        # Returns: False
        ```

    Example: Create an existing table without ignore_if_exists
        ```python
        # Try to create an existing table with ignore_if_exists=False
        session.catalog.create_table('my_table', Schema([
            ColumnField('id', IntegerType),
        ]), ignore_if_exists=False)
        # Raises: TableAlreadyExistsError
        ```
    """
    return self.catalog.create_table(table_name, schema, ignore_if_exists)

describe_table

describe_table(table_name: str) -> Schema

Returns the schema of the specified table.

Parameters:

• table_name (str) –

  Fully qualified or relative table name to describe.

Returns:

• Schema ( Schema ) –

  A schema object describing the table's structure with field names and types.

Raises:

• TableNotFoundError –

  If the table doesn't exist.

Describe a table's schema

# For a table created with: CREATE TABLE t1 (id int)
session.catalog.describe_table('t1')
# Returns: Schema([
#     ColumnField('id', IntegerType),
# ])

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def describe_table(self, table_name: str) -> Schema:
    """Returns the schema of the specified table.

    Args:
        table_name (str): Fully qualified or relative table name to describe.

    Returns:
        Schema: A schema object describing the table's structure with field names and types.

    Raises:
        TableNotFoundError: If the table doesn't exist.

    Example: Describe a table's schema
        ```python
        # For a table created with: CREATE TABLE t1 (id int)
        session.catalog.describe_table('t1')
        # Returns: Schema([
        #     ColumnField('id', IntegerType),
        # ])
        ```
    """
    return self.catalog.describe_table(table_name)

does_catalog_exist

does_catalog_exist(catalog_name: str) -> bool

Checks if a catalog with the specified name exists.

Parameters:

• catalog_name (str) –

  Name of the catalog to check.

Returns:

• bool ( bool ) –

  True if the catalog exists, False otherwise.

Check if a catalog exists

# Check if 'my_catalog' exists
session.catalog.does_catalog_exist('my_catalog')
# Returns: True

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def does_catalog_exist(self, catalog_name: str) -> bool:
    """Checks if a catalog with the specified name exists.

    Args:
        catalog_name (str): Name of the catalog to check.

    Returns:
        bool: True if the catalog exists, False otherwise.

    Example: Check if a catalog exists
        ```python
        # Check if 'my_catalog' exists
        session.catalog.does_catalog_exist('my_catalog')
        # Returns: True
        ```
    """
    return self.catalog.does_catalog_exist(catalog_name)

does_database_exist

does_database_exist(database_name: str) -> bool

Checks if a database with the specified name exists.

Parameters:

• database_name (str) –

  Fully qualified or relative database name to check.

Returns:

• bool ( bool ) –

  True if the database exists, False otherwise.

Check if a database exists

# Check if 'my_database' exists
session.catalog.does_database_exist('my_database')
# Returns: True

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def does_database_exist(self, database_name: str) -> bool:
    """Checks if a database with the specified name exists.

    Args:
        database_name (str): Fully qualified or relative database name to check.

    Returns:
        bool: True if the database exists, False otherwise.

    Example: Check if a database exists
        ```python
        # Check if 'my_database' exists
        session.catalog.does_database_exist('my_database')
        # Returns: True
        ```
    """
    return self.catalog.does_database_exist(database_name)

does_table_exist

does_table_exist(table_name: str) -> bool

Checks if a table with the specified name exists.

Parameters:

• table_name (str) –

  Fully qualified or relative table name to check.

Returns:

• bool ( bool ) –

  True if the table exists, False otherwise.

Check if a table exists

# Check if 'my_table' exists
session.catalog.does_table_exist('my_table')
# Returns: True

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def does_table_exist(self, table_name: str) -> bool:
    """Checks if a table with the specified name exists.

    Args:
        table_name (str): Fully qualified or relative table name to check.

    Returns:
        bool: True if the table exists, False otherwise.

    Example: Check if a table exists
        ```python
        # Check if 'my_table' exists
        session.catalog.does_table_exist('my_table')
        # Returns: True
        ```
    """
    return self.catalog.does_table_exist(table_name)

drop_catalog

drop_catalog(catalog_name: str, ignore_if_not_exists: bool = True) -> bool

Drops a catalog.

Parameters:

• catalog_name (str) –

  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 if the catalog doesn't exist. Defaults to True.

Raises:

• CatalogNotFoundError –

  If the catalog does not exist and ignore_if_not_exists is False

Returns:

• bool ( bool ) –

  True if the catalog was dropped successfully, False if the catalog

• bool –

  didn't exist and ignore_if_not_exists is True.

Drop a non-existent catalog

# Try to drop a non-existent catalog
session.catalog.drop_catalog('my_catalog')
# Returns: False

Drop a non-existent catalog without ignore_if_not_exists

# Try to drop a non-existent catalog with ignore_if_not_exists=False
session.catalog.drop_catalog('my_catalog', ignore_if_not_exists=False)
# Raises: CatalogNotFoundError

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def drop_catalog(
    self, catalog_name: str, ignore_if_not_exists: bool = True
) -> bool:
    """Drops a catalog.

    Args:
        catalog_name (str): Name of the catalog to drop.
        ignore_if_not_exists (bool): If True, silently return if the catalog doesn't exist.
            If False, raise an error if the catalog doesn't exist.
            Defaults to True.

    Raises:
        CatalogNotFoundError: If the catalog does not exist and ignore_if_not_exists is False

    Returns:
        bool: True if the catalog was dropped successfully, False if the catalog
        didn't exist and ignore_if_not_exists is True.

    Example: Drop a non-existent catalog
        ```python
        # Try to drop a non-existent catalog
        session.catalog.drop_catalog('my_catalog')
        # Returns: False
        ```

    Example: Drop a non-existent catalog without ignore_if_not_exists
        ```python
        # Try to drop a non-existent catalog with ignore_if_not_exists=False
        session.catalog.drop_catalog('my_catalog', ignore_if_not_exists=False)
        # Raises: CatalogNotFoundError
        ```
    """
    return self.catalog.drop_catalog(catalog_name, ignore_if_not_exists)

drop_database

drop_database(database_name: str, cascade: bool = False, ignore_if_not_exists: bool = True) -> bool

Drops a database.

Parameters:

• database_name (str) –

  Fully qualified or relative database name to drop.

• cascade (bool, default: False ) –

  If True, drop all tables in the database. Defaults to False.

• ignore_if_not_exists (bool, default: True ) –

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

Raises:

• DatabaseNotFoundError –

  If the database does not exist and ignore_if_not_exists is False

• CatalogError –

  If the current database is being dropped, if the database is not empty and cascade is False

Returns:

• bool ( bool ) –

  True if the database was dropped successfully, False if the database

• bool –

  didn't exist and ignore_if_not_exists is True.

Drop a non-existent database

# Try to drop a non-existent database
session.catalog.drop_database('my_database')
# Returns: False

Drop a non-existent database without ignore_if_not_exists

# Try to drop a non-existent database with ignore_if_not_exists=False
session.catalog.drop_database('my_database', ignore_if_not_exists=False)
# Raises: DatabaseNotFoundError

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def drop_database(
    self,
    database_name: str,
    cascade: bool = False,
    ignore_if_not_exists: bool = True,
) -> bool:
    """Drops a database.

    Args:
        database_name (str): Fully qualified or relative database name to drop.
        cascade (bool): If True, drop all tables in the database.
            Defaults to False.
        ignore_if_not_exists (bool): If True, silently return if the database doesn't exist.
            If False, raise an error if the database doesn't exist.
            Defaults to True.

    Raises:
        DatabaseNotFoundError: If the database does not exist and ignore_if_not_exists is False
        CatalogError: If the current database is being dropped, if the database is not empty and cascade is False

    Returns:
        bool: True if the database was dropped successfully, False if the database
        didn't exist and ignore_if_not_exists is True.

    Example: Drop a non-existent database
        ```python
        # Try to drop a non-existent database
        session.catalog.drop_database('my_database')
        # Returns: False
        ```

    Example: Drop a non-existent database without ignore_if_not_exists
        ```python
        # Try to drop a non-existent database with ignore_if_not_exists=False
        session.catalog.drop_database('my_database', ignore_if_not_exists=False)
        # Raises: DatabaseNotFoundError
        ```
    """
    return self.catalog.drop_database(database_name, cascade, ignore_if_not_exists)

drop_table

drop_table(table_name: str, ignore_if_not_exists: bool = True) -> bool

Drops the specified table.

By default this method will return False if the table doesn't exist.

Parameters:

• table_name (str) –

  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 when the table doesn't exist. Defaults to True.

Returns:

• bool ( bool ) –

  True if the table was dropped successfully, False if the table

• bool –

  didn't exist and ignore_if_not_exist is True.

Raises:

• TableNotFoundError –

  If the table doesn't exist and ignore_if_not_exists is False

Drop an existing table

# Drop an existing table 't1'
session.catalog.drop_table('t1')
# Returns: True

Drop a non-existent table with ignore_if_not_exists

# Try to drop a non-existent table with ignore_if_not_exists=True
session.catalog.drop_table('t2', ignore_if_not_exists=True)
# Returns: False

Drop a non-existent table without ignore_if_not_exists

# Try to drop a non-existent table with ignore_if_not_exists=False
session.catalog.drop_table('t2', ignore_if_not_exists=False)
# Raises: TableNotFoundError

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def drop_table(self, table_name: str, ignore_if_not_exists: bool = True) -> bool:
    """Drops the specified table.

    By default this method will return False if the table doesn't exist.

    Args:
        table_name (str): Fully qualified or relative table name to drop.
        ignore_if_not_exists (bool): If True, return False when the table doesn't exist.
            If False, raise an error when the table doesn't exist.
            Defaults to True.

    Returns:
        bool: True if the table was dropped successfully, False if the table
        didn't exist and ignore_if_not_exist is True.

    Raises:
        TableNotFoundError: If the table doesn't exist and ignore_if_not_exists is False

    Example: Drop an existing table
        ```python
        # Drop an existing table 't1'
        session.catalog.drop_table('t1')
        # Returns: True
        ```

    Example: Drop a non-existent table with ignore_if_not_exists
        ```python
        # Try to drop a non-existent table with ignore_if_not_exists=True
        session.catalog.drop_table('t2', ignore_if_not_exists=True)
        # Returns: False
        ```

    Example: Drop a non-existent table without ignore_if_not_exists
        ```python
        # Try to drop a non-existent table with ignore_if_not_exists=False
        session.catalog.drop_table('t2', ignore_if_not_exists=False)
        # Raises: TableNotFoundError
        ```
    """
    return self.catalog.drop_table(table_name, ignore_if_not_exists)

get_current_catalog

get_current_catalog() -> str

Returns the name of the current catalog.

Returns:

• str ( str ) –

  The name of the current catalog.

Get current catalog name

# Get the name of the current catalog
session.catalog.get_current_catalog()
# Returns: 'default'

Source code in src/fenic/api/catalog.py

def get_current_catalog(self) -> str:
    """Returns the name of the current catalog.

    Returns:
        str: The name of the current catalog.

    Example: Get current catalog name
        ```python
        # Get the name of the current catalog
        session.catalog.get_current_catalog()
        # Returns: 'default'
        ```
    """
    return self.catalog.get_current_catalog()

get_current_database

get_current_database() -> str

Returns the name of the current database in the current catalog.

Returns:

• str ( str ) –

  The name of the current database.

Get current database name

# Get the name of the current database
session.catalog.get_current_database()
# Returns: 'default'

Source code in src/fenic/api/catalog.py

def get_current_database(self) -> str:
    """Returns the name of the current database in the current catalog.

    Returns:
        str: The name of the current database.

    Example: Get current database name
        ```python
        # Get the name of the current database
        session.catalog.get_current_database()
        # Returns: 'default'
        ```
    """
    return self.catalog.get_current_database()

list_catalogs

list_catalogs() -> List[str]

Returns a list of available catalogs.

Returns:

• List[str] –

  List[str]: A list of catalog names available in the system.

• List[str] –

  Returns an empty list if no catalogs are found.

List all catalogs

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

Source code in src/fenic/api/catalog.py

def list_catalogs(self) -> List[str]:
    """Returns a list of available catalogs.

    Returns:
        List[str]: A list of catalog names available in the system.
        Returns an empty list if no catalogs are found.

    Example: List all catalogs
        ```python
        # Get all available catalogs
        session.catalog.list_catalogs()
        # Returns: ['default', 'my_catalog', 'other_catalog']
        ```
    """
    return self.catalog.list_catalogs()

list_databases

list_databases() -> List[str]

Returns a list of databases in the current catalog.

Returns:

• List[str] –

  List[str]: A list of database names in the current catalog.

• List[str] –

  Returns an empty list if no databases are found.

List all databases

# Get all databases in the current catalog
session.catalog.list_databases()
# Returns: ['default', 'my_database', 'other_database']

Source code in src/fenic/api/catalog.py

def list_databases(self) -> List[str]:
    """Returns a list of databases in the current catalog.

    Returns:
        List[str]: A list of database names in the current catalog.
        Returns an empty list if no databases are found.

    Example: List all databases
        ```python
        # Get all databases in the current catalog
        session.catalog.list_databases()
        # Returns: ['default', 'my_database', 'other_database']
        ```
    """
    return self.catalog.list_databases()

list_tables

list_tables() -> List[str]

Returns a list of tables stored in the current database.

This method queries the current database to retrieve all available table names.

Returns:

• List[str] –

  List[str]: A list of table names stored in the database.

• List[str] –

  Returns an empty list if no tables are found.

List all tables

# Get all tables in the current database
session.catalog.list_tables()
# Returns: ['table1', 'table2', 'table3']

Source code in src/fenic/api/catalog.py

def list_tables(self) -> List[str]:
    """Returns a list of tables stored in the current database.

    This method queries the current database to retrieve all available table names.

    Returns:
        List[str]: A list of table names stored in the database.
        Returns an empty list if no tables are found.

    Example: List all tables
        ```python
        # Get all tables in the current database
        session.catalog.list_tables()
        # Returns: ['table1', 'table2', 'table3']
        ```
    """
    return self.catalog.list_tables()

set_current_catalog

set_current_catalog(catalog_name: str) -> None

Sets the current catalog.

Parameters:

• catalog_name (str) –

  Name of the catalog to set as current.

Raises:

• ValueError –

  If the specified catalog doesn't exist.

Set current catalog

# Set 'my_catalog' as the current catalog
session.catalog.set_current_catalog('my_catalog')

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def set_current_catalog(self, catalog_name: str) -> None:
    """Sets the current catalog.

    Args:
        catalog_name (str): Name of the catalog to set as current.

    Raises:
        ValueError: If the specified catalog doesn't exist.

    Example: Set current catalog
        ```python
        # Set 'my_catalog' as the current catalog
        session.catalog.set_current_catalog('my_catalog')
        ```
    """
    self.catalog.set_current_catalog(catalog_name)

set_current_database

set_current_database(database_name: str) -> None

Sets the current database.

Parameters:

• database_name (str) –

  Fully qualified or relative database name to set as current.

Raises:

• DatabaseNotFoundError –

  If the specified database doesn't exist.

Set current database

# Set 'my_database' as the current database
session.catalog.set_current_database('my_database')

Source code in src/fenic/api/catalog.py

@validate_call(config=ConfigDict(strict=True))
def set_current_database(self, database_name: str) -> None:
    """Sets the current database.

    Args:
        database_name (str): Fully qualified or relative database name to set as current.

    Raises:
        DatabaseNotFoundError: If the specified database doesn't exist.

    Example: Set current database
        ```python
        # Set 'my_database' as the current database
        session.catalog.set_current_database('my_database')
        ```
    """
    self.catalog.set_current_database(database_name)