pyfabricops

https://pyfabricops.readthedocs.io/en/latest/
Admin
pyfabricops is a Python wrapper library for Microsoft Fabric and Power BI operations, providing a...
Tokens:152,411
Snippets:705
Trust Score:9.6
Update:2 months ago
Show doc for...
Tokens:
Raw
### Basic Usage Example

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging_system

Standard initialization for library operations.

```python
import pyfabricops as pf

# Default setup
pf.setup_logging(level='INFO')

# Use library normally
workspaces = pf.list_workspaces()
```

--------------------------------

### List domains usage examples

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/domains

Demonstrates how to retrieve all domains or filter for non-empty domains.

```python
list_domains() # List all domains
list_domains(non_empty_only=True) # List only non-empty domains
```

--------------------------------

### Create SQL On-Premises Connection Example

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/connections

Demonstrates how to initialize the connection using environment variables for credentials.

```python
from dotenv import load_dotenv
load_dotenv()

create_sql_on_premises_connection(
    display_name='My SQL On-Premises Connection',
    gateway_id='123e4567-e89b-12d3-a456-426614174000',
    server='myserver.database.windows.net',
    database='mydatabase',
    username=os.getenv('SQL_USERNAME'),
    password=os.getenv('SQL_PASSWORD'),
    credential_type='Basic',
    privacy_level='Organizational',
    connection_encryption='NotEncrypted',
    skip_test_connection=False,
    df=True,
)
```

--------------------------------

### Install pyfabricops

Source: https://pyfabricops.readthedocs.io/en/latest

Install the library using pip.

```bash
pip install -U pyfabricops
```

--------------------------------

### Define Environment Variables

Source: https://pyfabricops.readthedocs.io/en/latest

Example content for a .env file containing required authentication credentials.

```text
FAB_CLIENT_ID=your_client_id_here
FAB_CLIENT_SECRET=your_client_secret_here
FAB_TENANT_ID=your_tenant_id_here
FAB_USERNAME=your_username_here   # Necessary for some functions with no SPN support
FAB_PASSWORD=your_password_here   # Necessary for some functions with no SPN support
```

--------------------------------

### Publishing and Testing with TestPyPI

Source: https://pyfabricops.readthedocs.io/en/latest

Commands to configure, build, and publish the package to TestPyPI, as well as how to install it for verification.

```bash
# Configure TestPyPI
poetry config repositories.testpypi https://test.pypi.org/legacy/
poetry config pypi-token.testpypi <your-testpypi-token>

# Build and publish to TestPyPI
poetry build
poetry publish -r testpypi

# Install from TestPyPI
pip install --index-url https://test.pypi.org/simple/ pyfabricops
```

--------------------------------

### Log Output Format Example

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging_system

Sample output showing timestamp, module, level, and message structure.

```text
13:00:25 | pyfabricops._core    | INFO     | Making GET request to https://api.fabric.microsoft.com/v1/workspaces
13:00:25 | pyfabricops._core    | DEBUG    | Headers: {'Content-Type': 'application/json', 'Authorization': 'Bearer ***'}
13:00:25 | pyfabricops._core    | DEBUG    | Response status: 200
```

--------------------------------

### Update Notebook Definition Examples

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/notebooks

Demonstrates updating a notebook by name or ID with new display names or descriptions.

```python
update_notebook('MyProjectWorkspace', 'SalesDataModel', display_name='UpdatedSalesDataModel')
update_notebook('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', description='Updated description')
```

--------------------------------

### Get All Notebooks Configuration

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/notebooks

Retrieves a dictionary of configurations for all notebooks within a specified workspace.

```python
def get_all_notebooks_config(
    workspace: str,
) -> Union[Dict[str, Any], None]:
    """
    Get notebooks config from a workspace.

    Args:
        workspace (str): The name or ID from the workspace.

    Returns:
        (Union[Dict[str, Any], None]): The dict config of all notebooks in the workspace
    """
    items = list_notebooks(workspace, df=False)

    if items is None:
        return None

    config = {}

    for item in items:

        item_data = get_notebook(workspace, item['id'], df=False)

        config[item['displayName']] = {
            'id': item['id'],
            'description': item.get('description', None),
            'folder_id': ''
            if item.get('folderId') is None or pd.isna(item.get('folderId'))
            else item['folderId'],
        }

    return config
```

--------------------------------

### Update Dataflow Definition Examples

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/dataflows_gen2

Usage examples for updating a dataflow by name or ID.

```python
update_dataflow('MyProjectWorkspace', 'SalesDataModel', display_name='UpdatedSalesDataModel')
update_dataflow('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', description='Updated description')
```

--------------------------------

### Configure Logging System

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging_system

Basic setup and management of the logging system, including debug mode and reset functionality.

```python
import pyfabricops as pf

# Basic configuration
pf.setup_logging(level='INFO', format_style='standard')

# Debug mode for development
pf.enable_debug_mode(include_external=False)

# Disable logging completely
pf.disable_logging()

# Reset to default configuration
pf.reset_logging()
```

--------------------------------

### Retrieve workspace details examples

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/workspaces

Demonstrates how to call the workspace retrieval function using either a UUID or a workspace name.

```python
get_workspace_details('123e4567-e89b-12d3-a456-426614174000')
get_workspace_details('MyProject')
```

--------------------------------

### Get All Reports Configuration

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/reports

Retrieves a dictionary containing configuration details for all reports in a given workspace.

```python
def get_all_reports_config(
    workspace: str,
) -> Union[Dict[str, Any], None]:
    """
    Get reports config from a workspace.

    Args:
        workspace (str): The name or ID from the workspace.

    Returns:
        (Union[Dict[str, Any], None]): The dict config of all reports in the workspace
    """
    items = list_reports(workspace, df=False)

    if items is None:
        return None

    config = {}

    for item in items:

        item_data = get_report(workspace, item['id'], df=False)

        config[item['displayName']] = {
            'id': item['id'],
            'description': item.get('description', None),
            'folder_id': ''
            if item.get('folderId') is None or pd.isna(item.get('folderId'))
            else item['folderId'],
        }

    return config
```

--------------------------------

### Update workspace properties

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/workspaces

Examples demonstrating how to update a workspace's display name, description, or both using its ID or name.

```python
update_workspace(
    '123e4567-e89b-12d3-a456-426614174000',
    display_name='New Workspace Name',
)
```

```python
update_workspace(
    'MyProjectWorkspace',
    description='Updated description',
)
```

```python
update_workspace(
    '123e4567-e89b-12d3-a456-426614174000',
    display_name='New Workspace Name',
    description='Updated description',
)
```

--------------------------------

### Get All Lakehouses Configuration

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/lakehouses

Retrieves a configuration dictionary for all valid lakehouses in a workspace, including SQL endpoint details.

```python
def get_all_lakehouses_config(workspace: str) -> Union[Dict[str, Any], None]:
    """
    Generate lakehouses config from a workspace.

    Args:
        workspace (str): The name or ID from the workspace.

    Returns:
        (Union[Dict[str, Any], None]): The dict config from the lakehouses of the workspace
    """
    items = list_valid_lakehouses(workspace, df=False)

    if items is None:
        return None

    config = {}

    for item in items:

        item_data = get_lakehouse(workspace, item['id'], df=False)

        config[item['displayName']] = {
            'id': item['id'],
            'description': item.get('description', None),
            'folder_id': ''
            if item.get('folderId') is None or pd.isna(item.get('folderId'))
            else item['folderId'],
            'sql_endpoint_connection_string': item_data['properties'][
                'sqlEndpointProperties'
            ]['connectionString'],
            'sql_endpoint_id': item_data['properties'][
                'sqlEndpointProperties'
            ]['id'],
        }

    return config
```

--------------------------------

### Perform GET request to capacities endpoint

Source: https://pyfabricops.readthedocs.io/en/latest/functions/api/api

Executes a standard GET request to the Microsoft Fabric capacities endpoint.

```python
_base_api('capacities')
```

--------------------------------

### __init__(include_colors=True, include_module=True, ultra_minimal=False, include_symbols=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging

Initializes the custom logging formatter with specific display preferences.

```APIDOC
## __init__(include_colors=True, include_module=True, ultra_minimal=False, include_symbols=True)

### Description
Initializes the custom formatter for log records.

### Parameters
- **include_colors** (bool) - Optional - Whether to include colors in console output. Default: True
- **include_module** (bool) - Optional - Whether to include module name in log format. Default: True
- **ultra_minimal** (bool) - Optional - Whether to use ultra minimal format (just message). Default: False
- **include_symbols** (bool) - Optional - Whether to include colored symbols before messages. Default: True
```

--------------------------------

### deploy_notebook(workspace, path, start_path=None, description=None, df=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/notebooks

Deploys a single notebook to the target workspace.

```APIDOC
## deploy_notebook(workspace, path, start_path=None, description=None, df=True)

### Description
Deploy a single notebook to a specified workspace.

### Parameters
- **workspace** (str) - Required - The name or ID of the workspace.
- **path** (str) - Required - The path to the notebook.
- **start_path** (Optional[str]) - Optional - The starting path for folder creation.
- **description** (Optional[str]) - Optional - Description for the notebook.
- **df** (Optional[bool]) - Optional - If True, returns a DataFrame, otherwise returns a dictionary.

### Returns
- **Union[DataFrame, Dict[str, Any], None]** - The deployed notebook or None if deployment fails.
```

--------------------------------

### Get Token from Cache

Source: https://pyfabricops.readthedocs.io/en/latest/functions/api/auth

Retrieves a specific token dictionary by its key.

```python
def get_token(self, token_key: str) -> Optional[Dict]:
    """Get a specific token from cache"""
    tokens = self.load_tokens()
    return tokens.get(token_key)
```

--------------------------------

### Configure Logging with setup_logging

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging

Initializes the pyfabricops logger with specific formatting, rotation, and output settings.

```python
def setup_logging(
    level: Union[str, int] = logging.INFO,
    format_style: Literal['standard', 'minimal', 'detailed'] = 'standard',
    include_colors: bool = True,
    include_module: bool = True,
    include_symbols: bool = True,
    log_file: Optional[Union[str, Path]] = None,
    include_external: bool = False,
    max_file_size: int = 10 * 1024 * 1024,  # 10MB
    backup_count: int = 5,
) -> None:
    """
    Configure logging for pyfabricops.

    Args:
        level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL or numeric)
        format_style: Format style ('standard', 'minimal', 'detailed')
                     - 'minimal': Only message (ultra-minimal)
                     - 'standard': Timestamp, level, and message
                     - 'detailed': Timestamp, module, level, and message
        include_colors: Whether to use colored output in console
        include_module: Whether to include module names in logs
        include_symbols: Whether to include colored symbols before messages (enabled by default)
        log_file: Optional file path to write logs to
        include_external: Whether to include logs from external libraries
        max_file_size: Maximum size of log file before rotation (bytes)
        backup_count: Number of backup files to keep when rotating
    """
    # Convert string level to numeric if needed
    if isinstance(level, str):
        level = getattr(logging, level.upper(), logging.INFO)

    # Get the root logger for pyfabricops
    root_logger = logging.getLogger('pyfabricops')

    # Clear existing handlers to avoid duplicates
    for handler in root_logger.handlers[:]:
        root_logger.removeHandler(handler)

    # Set the logging level
    root_logger.setLevel(level)

    # Create console handler
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(level)

    # Create formatter based on style
    if format_style == 'minimal':
        formatter = PyFabricOpsFormatter(
            include_colors=include_colors,
            include_module=False,
            ultra_minimal=True,
            include_symbols=include_symbols,
        )
    elif format_style == 'detailed':
        formatter = PyFabricOpsFormatter(
            include_colors=include_colors,
            include_module=True,
            ultra_minimal=False,
            include_symbols=include_symbols,
        )
    else:  # standard
        formatter = PyFabricOpsFormatter(
            include_colors=include_colors,
            include_module=include_module,
            ultra_minimal=False,
            include_symbols=include_symbols,
        )

    console_handler.setFormatter(formatter)

    # Add filter
    console_filter = PyFabricOpsFilter(include_external=include_external)
    console_handler.addFilter(console_filter)

    # Add console handler
    root_logger.addHandler(console_handler)

    # Add file handler if specified
    if log_file:
        log_path = Path(log_file)
        log_path.parent.mkdir(parents=True, exist_ok=True)
```

--------------------------------

### Usage of df decorator

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/decorators

Example of calling a function with the df parameter set to True.

```python
list_capacities(df=True)
```

--------------------------------

### Get project root path

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/utils

Returns the current working directory as the root path.

```python
def get_root_path() -> Path:
    """
    Find the root path.
    """
    return os.getcwd()
```

--------------------------------

### Create a deployment pipeline

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/deployment_pipelines

Initializes a new deployment pipeline with specified stages and description. Returns a DataFrame by default when df is set to True.

```python
create_deployment_pipeline(
    display_name='My Deployment Pipeline',
    stages=[
        {
            "displayName": "Development",
            "description": "Development stage description",
            "isPublic": false
        },
        {
            "displayName": "Test",
            "description": "Test stage description",
            "isPublic": false
        },
        {
            "displayName": "Production",
            "description": "Production stage description",
            "isPublic": true
        }
    ],
    description='This is my deployment pipeline',
    df=True
)
```

```python
@df
def create_deployment_pipeline(
    display_name: str,
    stages: List[Dict[str, Any]],
    *,
    description: Optional[str] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    Creates a new deployment pipeline.

    Args:
        display_name (str): The display name of the deployment pipeline.
        stages (List[Dict[str, Any]]): A list of stages for the deployment pipeline.
        description (Optional[str]): An optional description for the deployment pipeline.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): The details of the created deployment pipeline.

    Examples:
        ```python
        create_deployment_pipeline(
            display_name='My Deployment Pipeline',
            stages=[
                {
                    "displayName": "Development",
                    "description": "Development stage description",
                    "isPublic": false
                },
                {
                    "displayName": "Test",
                    "description": "Test stage description",
                    "isPublic": false
                },
                {
                    "displayName": "Production",
                    "description": "Production stage description",
                    "isPublic": true
                }
            ],
            description='This is my deployment pipeline',
            df=True
        )
        ```
    """
    payload = {
        'displayName': display_name,
        'stages': stages,
    }
    if description:
        payload['description'] = description
    return api_request(
        '/deploymentPipelines',
        method='POST',
        payload=payload,
    )
```

--------------------------------

### Create support files with create_support_files

Source: https://pyfabricops.readthedocs.io/en/latest/functions/cd/support_files

This function iterates through a list of file definitions, creates the necessary directory structures, and writes the content to each specified path.

```python
def create_support_files():
    """
    Create support files with predefined content for PyFabricOps CI/CD operations.
    """
    files = [
        ENV,
        BRANCHES,
        WORKSPACES_ROLES,
        CONNECTIONS_ROLES,
        GITIGNORE,
        GITATTRIBUTES,
        SRC,
    ]

    # Create directories and files
    for file_dict in files:
        path = file_dict.get('path')
        if path:
            os.makedirs(os.path.dirname(path) or '.', exist_ok=True)
            with open(path, 'w', encoding='utf-8') as f:
                f.write(file_dict['content'])
            logger.success(f'Created {path}')
```

--------------------------------

### deploy_all_items(workspace, path, start_path=None)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/items

Deploys all items to a specified workspace.

```APIDOC
## deploy_all_items(workspace, path, start_path=None)

### Description
Deploys all items to the specified workspace.

### Parameters
- **workspace** (str) - Required - The name or ID of the workspace.
- **path** (str) - Required - The path to the notebooks.
- **start_path** (Optional[str]) - Optional - The starting path for folder creation. Defaults to None.
```

--------------------------------

### Update a domain

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/domains

Example usage of the update_domain function to modify a domain's display name.

```python
update_domain(
    domain_id='98f6b7c8-1234-5678-90ab-cdef12345678',
    display_name='NewdomainName',
)
```

```python
update_domain(
    domain_id='98f6b7c8-1234-5678-90ab-cdef12345678',
    display_name='NewdomainName',
)
```

--------------------------------

### Get a logger instance

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging

Retrieves a configured logger instance, ensuring the name is prefixed with pyfabricops.

```python
def get_logger(name: str) -> logging.Logger:
    """
    Get a logger instance for a specific module.

    Args:
        name: The logger name (usually __name__)

    Returns:
        logging.Logger: Configured logger instance
    """
    # Ensure the name starts with pyfabricops
    if not name.startswith('pyfabricops'):
        if name == '__main__':
            name = 'pyfabricops.main'
        else:
            name = f'pyfabricops.{name}'

    logger = logging.getLogger(name)

    # Add NullHandler as fallback if no configuration is set
    if not logger.handlers and not logger.parent.handlers:
        logger.addHandler(logging.NullHandler())

    return logger
```

--------------------------------

### git_init(workspace, *, initialize_strategy='PreferWorkspace', credential_type='spn', df=False)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/git

Initializes a Fabric workspace to use Git for version control.

```APIDOC
## git_init(workspace, *, initialize_strategy='PreferWorkspace', credential_type='spn', df=False)

### Description
Initializes a Fabric workspace to use Git for version control.

### Parameters
- **workspace** (str) - Required - The name or ID of the Fabric workspace.
- **initialize_strategy** (Literal['PreferWorkspace', 'PreferRemote', 'None']) - Optional - The strategy to use for initialization. Defaults to 'PreferWorkspace'.
- **credential_type** (Literal['spn', 'user']) - Optional - The type of credentials to use for the Git connection. Defaults to 'spn'.
- **df** (Optional[bool]) - Optional - If True, returns a DataFrame with flattened keys. If False, returns a list of dictionaries. Defaults to False.

### Returns
- **Union[DataFrame, Dict[str, Any], None]** - The response data from the API if successful, or None if the workspace cannot be resolved.

### Raises
- **ValueError** - If the workspace cannot be resolved.

### Example
```python
git_init(workspace='my_workspace', initialize_strategy='PreferWorkspace', provider='GitHub')
```
```

--------------------------------

### Extract middle path

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/utils

Extracts a segment of a file path, optionally relative to a start path.

```python
def extract_middle_path(
    path: str, start_path: Optional[str] = None
) -> Union[str, None]:
    """
    Extract the middle of a full path given a start.
    """
    path_list = path.split('/')[:-1]
    if len(path_list) == 0:
        return None

    middle_path = '/'.join(path_list)

    if start_path is None:
        return middle_path
    else:
        try:
            middle_path = middle_path.split(start_path + '/')[1]
            return middle_path
        except:
            return None
```

--------------------------------

### git_init(workspace, *, initialize_strategy='PreferWorkspace', credential_type='spn', df=False)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/git

Initializes a Fabric workspace to use Git for version control.

```APIDOC
## git_init(workspace, *, initialize_strategy='PreferWorkspace', credential_type='spn', df=False)

### Description
Initializes a Fabric workspace to use Git for version control.

### Parameters
- **workspace** (str) - Required - The name or ID of the Fabric workspace.
- **initialize_strategy** (Literal['PreferWorkspace', 'PreferRemote', 'None']) - Optional - The strategy to use for initialization. Defaults to 'PreferWorkspace'.
- **credential_type** (Literal['spn', 'user']) - Optional - The type of credentials to use for the Git connection. Defaults to 'spn'.
- **df** (Optional[bool]) - Optional - If True, returns a DataFrame with flattened keys. If False, returns a list of dictionaries. Defaults to False.

### Returns
- (Union[DataFrame, Dict[str, Any], None]) - The response data from the API if successful, or None if the workspace cannot be resolved.

### Raises
- ValueError: If the workspace cannot be resolved.
```

--------------------------------

### create_deployment_pipeline(display_name, stages, *, description=None, df=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/deployment_pipelines

Creates a new deployment pipeline with the specified stages and configuration.

```APIDOC
## create_deployment_pipeline(display_name, stages, *, description=None, df=True)

### Description
Creates a new deployment pipeline. Returns the details of the created pipeline as a DataFrame or list of dictionaries depending on the `df` parameter.

### Parameters
- **display_name** (str) - Required - The display name of the deployment pipeline.
- **stages** (List[Dict[str, Any]]) - Required - A list of stages for the deployment pipeline.
- **description** (Optional[str]) - Optional - An optional description for the deployment pipeline.
- **df** (Optional[bool]) - Optional - If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

### Returns
- **Union[DataFrame, List[Dict[str, Any]], None]** - The details of the created deployment pipeline.

### Example
```python
create_deployment_pipeline(
    display_name='My Deployment Pipeline',
    stages=[
        {"displayName": "Development", "description": "Development stage description", "isPublic": false},
        {"displayName": "Test", "description": "Test stage description", "isPublic": false},
        {"displayName": "Production", "description": "Production stage description", "isPublic": true}
    ],
    description='This is my deployment pipeline',
    df=True
)
```
```

--------------------------------

### deploy_all_notebooks(workspace, path, start_path=None)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/notebooks

Deploys all notebooks found at the specified path to the target workspace.

```APIDOC
## deploy_all_notebooks(workspace, path, start_path=None)

### Description
Deploy all notebooks to a specified workspace.

### Parameters
- **workspace** (str) - Required - The name or ID of the workspace.
- **path** (str) - Required - The path to the notebooks.
- **start_path** (Optional[str]) - Optional - The starting path for folder creation.
```

--------------------------------

### setup_logging

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/logging

Configures the logging system for pyfabricops, allowing customization of log levels, output formats, and file rotation policies.

```APIDOC
## setup_logging(level, format_style, include_colors, include_module, include_symbols, log_file, include_external, max_file_size, backup_count)

### Description
Configures the logging system for pyfabricops. This function sets the logging level, defines the output format, and optionally configures file-based logging with rotation.

### Parameters
- **level** (Union[str, int]) - Optional - Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL or numeric). Default: INFO
- **format_style** (Literal['standard', 'minimal', 'detailed']) - Optional - Format style. 'minimal': Only message; 'standard': Timestamp, level, and message; 'detailed': Timestamp, module, level, and message. Default: 'standard'
- **include_colors** (bool) - Optional - Whether to use colored output in console. Default: True
- **include_module** (bool) - Optional - Whether to include module names in logs. Default: True
- **include_symbols** (bool) - Optional - Whether to include colored symbols before messages. Default: True
- **log_file** (Optional[Union[str, Path]]) - Optional - Optional file path to write logs to. Default: None
- **include_external** (bool) - Optional - Whether to include logs from external libraries. Default: False
- **max_file_size** (int) - Optional - Maximum size of log file before rotation (bytes). Default: 10 * 1024 * 1024
- **backup_count** (int) - Optional - Number of backup files to keep when rotating. Default: 5
```

--------------------------------

### Get All Data Pipelines Configuration

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/data_pipelines

Retrieves a dictionary of configurations for all data pipelines within a specified workspace.

```python
def get_all_data_pipelines_config(
    workspace: str,
) -> Union[Dict[str, Any], None]:
    """
    Get data_pipelines config from a workspace.

    Args:
        workspace (str): The name or ID from the workspace.

    Returns:
        (Union[Dict[str, Any], None]): The dict config of all data_pipelines in the workspace
    """
    items = list_data_pipelines(workspace, df=False)

    if items is None:
        return None

    config = {}

    for item in items:

        item_data = get_data_pipeline(workspace, item['id'], df=False)

        config[item['displayName']] = {
            'id': item['id'],
            'description': item.get('description', None),
            'folder_id': ''
            if item.get('folderId') is None or pd.isna(item.get('folderId'))
            else item['folderId'],
        }

    return config
```

--------------------------------

### Update Warehouse Properties

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/warehouses

Examples demonstrating how to update the display name or description of a warehouse using its name or ID.

```python
update_warehouse('MyProjectWorkspace', 'SalesDatawarehouse', display_name='UpdatedSalesDatawarehouse')
update_warehouse('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', description='Updated description')
```

--------------------------------

### Create a Notebook

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/notebooks

Creates a new notebook in a workspace using a display name and a definition file path.

```python
create_notebook('MyProjectWorkspace', 'SalesDataModel', 'path/to/definition.json')
create_notebook('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', 'path/to/definition.json')
```

```python
@df
def create_notebook(
    workspace: str,
    display_name: str,
    item_definition: str,
    *,
    description: Optional[str] = None,
    folder: Optional[str] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Creates a new notebook in the specified workspace.

    Args:
        workspace (str): The workspace name or ID.
        display_name (str): The display name of the notebook.
        description (str, optional): A description for the notebook.
        folder (str, optional): The folder to create the notebook in.
        path (str): The path to the notebook definition file.

    Returns:
        (dict): The created notebook details.

    Examples:
        ```python
        create_notebook('MyProjectWorkspace', 'SalesDataModel', 'path/to/definition.json')
        create_notebook('MyProjectWorkspace', '123e4567-e89b-12d3-a456-426614174000', 'path/to/definition.json')
        ```
    """
    workspace_id = resolve_workspace(workspace)

    payload = {'displayName': display_name, 'definition': item_definition}

    if description:
        payload['description'] = description

    if folder:
        folder_id = resolve_folder(workspace_id, folder)
        if folder_id:
            payload['folderId'] = folder_id

    return api_request(
        endpoint='/workspaces/' + workspace_id + '/notebooks',
        method='post',
        payload=payload,
        support_lro=True,
    )
```

--------------------------------

### create_domain(display_name, *, description=None, parent_domain=None, df=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/domains

Creates a new domain in the tenant.

```APIDOC
## create_domain(display_name, *, description=None, parent_domain=None, df=True)

### Description
Create a new domain in the tenant.

### Parameters
- **display_name** (str) - Required - The name of the domain to create.
- **description** (str) - Optional - The description of the domain to create.
- **parent_domain** (str) - Optional - The name or ID of the parent domain.
- **df** (Optional[bool]) - Optional - If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

### Returns
- **Union[DataFrame, Dict[str, Any], None]** - The created domain details if successful, otherwise None.
```

--------------------------------

### Update data pipeline definition example

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/data_pipelines

Basic usage of the update_data_pipeline_definition function to modify a pipeline's definition.

```python
update_data_pipeline_definition(
    workspace='MyProjectWorkspace',
    data_pipeline='SalesDataPipeline',
    item_definition={...} # The definition of the data_pipeline
)
```

```python
@df
def update_data_pipeline_definition(
    workspace: str, data_pipeline: str, item_definition: Dict[str, Any]
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Updates the definition of an existing data_pipeline in the specified workspace.
    If the data_pipeline does not exist, it returns None.

    Args:
        workspace (str): The workspace name or ID.
        data_pipeline (str): The name or ID of the data_pipeline to update.
        item_definition (Dict[str, Any]): The item_definition of the data_pipeline.
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The updated data_pipeline details if successful, otherwise None.

    Examples:
        ```python
        update_data_pipeline_definition(
            workspace='MyProjectWorkspace',
            data_pipeline='SalesDataPipeline',
            item_definition={...} # The definition of the data_pipeline
        )
        ```
    """
    workspace_id = resolve_workspace(workspace)

    data_pipeline_id = resolve_data_pipeline(workspace_id, data_pipeline)

    params = {'updateMetadata': True}
    payload = {'definition': item_definition}

    return api_request(
        endpoint='/workspaces/'
        + workspace_id
        + '/dataPipelines/'
        + data_pipeline_id,
        method='post',
        payload=payload,
        params=params,
    )
```

--------------------------------

### Get Dataflow Gen2 Configuration

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/dataflows_gen2

Retrieves the configuration dictionary for a specific Dataflow Gen2 item within a workspace.

```python
def get_dataflow_gen2_config(
    workspace: str, dataflow_gen2: str
) -> Union[Dict[str, Any], None]:
    """
    Get a specific dataflow_gen2 config from a workspace.

    Args:
        workspace (str): The name or ID of the workspace.
        dataflow_gen2 (str): The name or ID of the dataflow_gen2.

    Returns:
        (Union[Dict[str, Any], None]): The dict config from the dataflow_gen2.
    """
    item = dataflow_gen2
    item_data = get_dataflow_gen2(workspace, item, df=False)

    if item_data is None:
        return None

    else:
        config = {}
        config = config[item_data.get('displayName')] = {}

        config = {
            'id': item_data['id'],
            'description': item_data.get('description', None),
            'folder_id': ''
            if item_data.get('folderId') is None
            or pd.isna(item_data.get('folderId'))
            else item_data['folderId'],
        }

        return config
```

--------------------------------

### create_workspace(display_name, *, capacity=None, description=None, df=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/workspaces

Creates a new workspace with the specified name, capacity, and description.

```APIDOC
## create_workspace(display_name, *, capacity=None, description=None, df=True)

### Description
Create a new workspace with the specified name, capacity and description.

### Parameters
- **display_name** (str) - Required - The name of the workspace to create.
- **capacity** (str) - Optional - The ID or name of the capacity to assign to the workspace. Defaults to None.
- **description** (str) - Optional - A description for the workspace. Defaults to None.
- **df** (Optional[bool]) - Optional - If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries.

### Returns
- **Union[DataFrame, Dict[str, Any], None]** - The details of the created or updated workspace if successful, otherwise None.
```

--------------------------------

### Get All Dataflow Gen2 Configurations

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/dataflows_gen2

Retrieves a dictionary of configurations for all Dataflow Gen2 items found in a specified workspace.

```python
def get_all_dataflows_gen2_config(
    workspace: str,
) -> Union[Dict[str, Any], None]:
    """
    Get dataflows_gen2 config from a workspace.

    Args:
        workspace (str): The name or ID from the workspace.

    Returns:
        (Union[Dict[str, Any], None]): The dict config of all dataflows_gen2 in the workspace
    """
    items = list_dataflows_gen2(workspace, df=False)

    if items is None:
        return None

    config = {}

    for item in items:

        item_data = get_dataflow_gen2(workspace, item['id'], df=False)

        config[item['displayName']] = {
            'id': item['id'],
            'description': item.get('description', None),
            'folder_id': ''
            if item.get('folderId') is None or pd.isna(item.get('folderId'))
            else item['folderId'],
        }

    return config
```

--------------------------------

### Get Lakehouse Configuration

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/lakehouses

Retrieves the configuration dictionary for a specific lakehouse identified by name or ID within a workspace.

```python
def get_lakehouse_config(
    workspace: str, lakehouse: str
) -> Union[Dict[str, Any], None]:
    """
    Get a specific lakehouse config from a workspace.

    Args:
        workspace (str): The name or ID from the workspace.
        lakehouse (str): The name or ID from the lakehouse.

    Returns:
        (Union[Dict[str, Any], None]): The dict config from the lakehouse
    """
    item = lakehouse
    item_data = get_lakehouse(workspace, item, df=False)

    if item_data is None:
        return None

    else:
        config = {}
        config = config[item_data.get('displayName')] = {}

        config = {
            'id': item_data['id'],
            'description': item_data.get('description', None),
            'folder_id': ''
            if item_data.get('folderId') is None
            or pd.isna(item_data.get('folderId'))
            else item_data['folderId'],
            'sql_endpoint_connection_string': item_data.get(
                'properties_sqlEndpointProperties_connectionString'
            ),
            'sql_endpoint_id': item_data.get(
                'properties_sqlEndpointProperties_id'
            ),
        }

        return config
```

--------------------------------

### create_notebook(workspace, display_name, item_definition, *, description=None, folder=None, df=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/notebooks

Creates a new notebook in the specified workspace.

```APIDOC
## create_notebook(workspace, display_name, item_definition, *, description=None, folder=None, df=True)

### Description
Creates a new notebook in the specified workspace.

### Parameters
- **workspace** (str) - Required - The workspace name or ID.
- **display_name** (str) - Required - The display name of the notebook.
- **item_definition** (str) - Required - The path to the notebook definition file.
- **description** (str) - Optional - A description for the notebook.
- **folder** (str) - Optional - The folder to create the notebook in.
- **df** (bool) - Optional - Default: True

### Returns
- **dict** - The created notebook details.

### Example
```python
create_notebook('MyProjectWorkspace', 'SalesDataModel', 'path/to/definition.json')
```
```

--------------------------------

### extract_middle_path(path, start_path=None)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/utils/utils

Extracts the middle portion of a full file path based on an optional start path.

```APIDOC
## extract_middle_path(path, start_path=None)

### Description
Extract the middle of a full path given a start.

### Parameters
- **path** (str) - Required - The full path string.
- **start_path** (str) - Optional - The starting path segment to strip.
```

--------------------------------

### create_support_files()

Source: https://pyfabricops.readthedocs.io/en/latest/functions/cd/support_files

Initializes and creates necessary support files for PyFabricOps CI/CD operations, including environment, branch, and workspace configurations.

```APIDOC
## create_support_files()

### Description
Creates support files with predefined content for PyFabricOps CI/CD operations. This function iterates through a predefined list of files and directories, ensuring the necessary file structure is created on the filesystem.

### Signature
`create_support_files()`

### Behavior
- Creates directories as needed.
- Writes predefined content to files including ENV, BRANCHES, WORKSPACES_ROLES, CONNECTIONS_ROLES, GITIGNORE, GITATTRIBUTES, and SRC.
- Logs success messages for each created file.
```

--------------------------------

### Initialize Git for a Fabric workspace

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/git

Sets up Git version control for a Fabric workspace using a specified initialization strategy.

```python
git_init(
    workspace='my_workspace',
    initialize_strategy='PreferWorkspace',
    provider='GitHub'
)
```

--------------------------------

### list_deployment_pipelines(df=True)

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/deployment_pipelines

Lists all available deployment pipelines.

```APIDOC
## list_deployment_pipelines(df=True)

### Description
List deployment pipelines. Returns either a DataFrame or a list of dictionaries based on the df parameter.

### Parameters
- **df** (Optional[bool]) - Optional - If True or not provided, returns a DataFrame with flattened keys. If False, returns a list of dictionaries. Defaults to True.

### Returns
- **Union[DataFrame, List[Dict[str, Any]], None]** - A list of deployment pipelines.

### Example
```python
list_deployment_pipelines()
```
```

--------------------------------

### Refresh a Semantic Model

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/semantic_models

Example usage of the refresh_semantic_model function with specific parameters for workspace, model, and object-level refresh configuration.

```python
refresh_semantic_model(
    workspace="AdventureWorks",
    semantic_model="SalesAnalysis",
    apply_refresh_policy=False,
    commit_mode="Transactional",
    effective_date="2023-01-01",
    max_parallelism=5,
    objects=[
        {
            "table": "FactSales",
            "partition": "2024"
        },
        {
            "table": "FactReturns",
            "partition": "2024"
        }
    ],
    retry_count=3,
    timeout="00:30:00",
    type="Full"
)
```

--------------------------------

### Update Workspace from Git

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/git

Example usage of the update_from_git function to synchronize a workspace with specific conflict resolution and override settings.

```python
update_from_git(
    workspace='my_workspace',
    conflict_resolution_policy='PreferRemote',
    allow_override_items=False
)
```

--------------------------------

### Deploy all notebooks

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/notebooks

Deploys all notebooks found at a specified path to the target workspace. It handles folder creation based on the provided path structure.

```python
def deploy_all_notebooks(
    workspace: str,
    path: str,
    start_path: Optional[str] = None,
) -> None:
    """
    Deploy all notebooks to workspace.

    Args:
        workspace (str): The name or ID of the workspace.
        path (str): The path to the notebooks.
        start_path (Optional[str]): The starting path for folder creation.
    """
    workspace_id = resolve_workspace(workspace)
    if workspace_id is None:
        return None

    notebooks_paths = list_paths_of_type(path, 'Notebook')

    for path_ in notebooks_paths:

        display_name = extract_display_name_from_platform(path_)
        if display_name is None:
            return None

        item_id = resolve_notebook(workspace_id, display_name)

        folder_path_string = extract_middle_path(path_, start_path=start_path)
        folder_id = create_folders_from_path_string(
            workspace_id, folder_path_string
        )

        item_definition = pack_item_definition(path_)

        if item_id is None:
            create_notebook(
                workspace_id,
                display_name=display_name,
                item_definition=item_definition,
                folder=folder_id,
                df=False,
            )

        else:
            update_notebook_definition(
                workspace_id,
                item_id,
                item_definition=item_definition,
                df=False,
            )

    logger.success(
        f'All notebooks were deployed to workspace "{workspace}" successfully.'
    )
    return None
```

--------------------------------

### Create SQL On-Premises Connection

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/connections

Configures and creates a new SQL connection via an on-premises gateway. Requires valid gateway credentials and server details.

```python
create_sql_on_premises_connection(
    display_name='My SQL On-Premises Connection',
    gateway_id='123e4567-e89b-12d3-a456-426614174000',
    server='myserver.database.windows.net',
    database='mydatabase',
    username=os.getenv('SQL_USERNAME'),
    password=os.getenv('SQL_PASSWORD'),
    credential_type='Basic',
    privacy_level='Organizational',
    connection_encryption='NotEncrypted',
    skip_test_connection=False,
    df=True,
)
```

```python
encrypted_credentials = _get_encrypt_gateway_credentials(
    gateway_id=gateway_id, username=username, password=password
)
payload = {
    'connectivityType': 'OnPremisesGateway',
    'gatewayId': gateway_id,
    'displayName': display_name,
    'connectionDetails': {
        'type': 'SQL',
        'creationMethod': 'SQL',
        'parameters': [
            {'dataType': 'Text', 'name': 'server', 'value': server},
            {'dataType': 'Text', 'name': 'database', 'value': database},
        ],
    },
    'privacyLevel': privacy_level,
    'credentialDetails': {
        'singleSignOnType': 'None',
        'connectionEncryption': connection_encryption,
        'skipTestConnection': skip_test_connection,
        'credentials': {
            'credentialType': credential_type,
            'values': [
                {
                    'gatewayId': gateway_id,
                    'encryptedCredentials': encrypted_credentials,
                }
            ],
        },
    },
}
return api_request(
    '/connections',
    method='post',
    payload=payload,
)
```

--------------------------------

### Update Dataflow Gen1 Definition Usage

Source: https://pyfabricops.readthedocs.io/en/latest/functions/items/dataflows_gen1

Example of calling the update_dataflow_gen1_definition function with workspace, dataflow name, and definition dictionary.

```python
update_dataflow_gen1_definition(
    workspace='MyProjectWorkspace',
    dataflow='SalesDataflowGen1',
    item_definition={...} # The definition of the dataflow
)
```

```python
update_dataflow_gen1_definition(
    workspace='MyProjectWorkspace',
    dataflow='SalesDataflowGen1',
    item_definition={...} # The definition of the dataflow
)
```

--------------------------------

### List deployment pipelines

Source: https://pyfabricops.readthedocs.io/en/latest/functions/core/deployment_pipelines

Retrieves a list of all available deployment pipelines.

```python
list_deployment_pipelines()
```

```python
@df
def list_deployment_pipelines(
    df: Optional[bool] = True,
) -> Union[DataFrame, List[Dict[str, Any]], None]:
    """
    List deployment pipelines

    Args:
        df (Optional[bool]): If True or not provided, returns a DataFrame with flattened keys.
            If False, returns a list of dictionaries.

    Returns:
        (Union[DataFrame, List[Dict[str, Any]], None]): A list of deployment pipelines.

    Examples:
        ```python
        list_deployment_pipelines()
        ```
    """
    return api_request(
        '/deploymentPipelines',
        support_pagination=True,
    )
```

--------------------------------

### Deploy a single notebook

Source: https://pyfabricops.readthedocs.io/en/latest/functions/helpers/notebooks

Deploys a single notebook to the workspace, returning either a DataFrame or a dictionary based on the df parameter.

```python
@df
def deploy_notebook(
    workspace: str,
    path: str,
    start_path: Optional[str] = None,
    description: Optional[str] = None,
    df: Optional[bool] = True,
) -> Union[DataFrame, Dict[str, Any], None]:
    """
    Deploy a notebook to workspace.

    Args:
        workspace (str): The name or ID of the workspace.
        path (str): The path to the notebook.
        start_path (Optional[str]): The starting path for folder creation.
        description (Optional[str]): Description for the notebook.
        df (Optional[bool]): If True, returns a DataFrame, otherwise returns a dictionary.

    Returns:
        (Union[DataFrame, Dict[str, Any], None]): The deployed notebook or None if deployment fails.
    """
    workspace_id = resolve_workspace(workspace)
    if workspace_id is None:
        return None

    display_name = extract_display_name_from_platform(path)
    if display_name is None:
        return None

    item_id = resolve_notebook(workspace_id, display_name)

    folder_path_string = extract_middle_path(path, start_path=start_path)
    folder_id = create_folders_from_path_string(
        workspace_id, folder_path_string
    )

    item_definition = pack_item_definition(path)

    if item_id is None:
        return create_notebook(
            workspace_id,
            display_name=display_name,
            item_definition=item_definition,
            description=description,
            folder=folder_id,
            df=False,
        )

    else:
        return update_notebook_definition(
            workspace_id,
            item_id,
            item_definition=item_definition,
            df=False,
        )
```
Tokens:
Raw