### 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 # 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, ) ```