This repository contains the Python SDK for Azure Connectors. Code must follow Python best practices (PEP 8, PEP 484) and the team's conventions for async/await patterns, type safety, and API design.
# ------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ------------------------------------------------------------
"""Module docstring describing the purpose of this module."""
from __future__ import annotations
import asyncio
from typing import Any, Optional
from azure.connectors.sdk import ConnectorClient
class YourClass:
"""Class docstring."""
passRules:
- Copyright header: Use
# ----(4 dashes) format with double space before "All rights reserved" - Module docstring immediately after copyright
- Use
from __future__ import annotationsfor forward references - Imports sorted: standard library, third-party, local (use
isort) - One blank line between import groups
| Element | Rule | Example |
|---|---|---|
| Modules | lowercase with underscores | http_client.py |
| Classes | PascalCase | ConnectorClient |
| Functions/methods | snake_case | get_connection_status() |
| Constants | UPPER_SNAKE_CASE | DEFAULT_TIMEOUT_SECONDS |
| Private members | leading underscore | _internal_method(), _private_attr |
| Type variables | PascalCase with suffix | ResponseT, ItemT |
Variable naming rules:
- Use complete, unabbreviated English terms for all identifiers
- No single-letter variable names except for well-known conventions (
i,jfor indices,efor exceptions in handlers) - No placeholder names (
blah,foo,temp,x) — always use meaningful names
One primary class per module:
- Main class should match the module name (e.g.,
http_client.pycontainsHttpClient) - Helper classes and functions can coexist in the same module if tightly coupled
- Keep modules focused — split large modules into submodules
ALWAYS use type hints for public APIs:
from typing import Optional
async def get_user(
user_id: str,
include_details: bool = False,
) -> Optional[User]:
"""Retrieve a user by ID."""
...Rules:
- Use
from __future__ import annotationsfor modern syntax - Use
Optional[X]orX | Nonefor nullable types - Use
list[str](lowercase) instead ofList[str](Python 3.9+) - Return type is required for all public functions/methods
- Use
TypedDictfor structured dictionaries - Use
Protocolfor duck typing instead of ABCs when appropriate
Use async context managers:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.json()Rules:
- Use
async withfor resources that need cleanup - Prefer
asyncio.gather()for concurrent operations - Never use
asyncio.run()inside async code - Use
asyncio.create_task()for fire-and-forget operations (with proper error handling)
DO NOT:
# Wrong: Blocking call in async function
response = requests.get(url) # Use aiohttp instead
# Wrong: Manual cleanup without context manager
session = aiohttp.ClientSession()
try:
...
finally:
await session.close() # Use async with insteadSingle line if all parameters fit:
def process_item(item: Item, validate: bool = True) -> Result:Multi-line for longer signatures:
async def create_connection(
connection_id: str,
connection_type: str,
*,
timeout_seconds: int = 30,
retry_count: int = 3,
) -> Connection:Rules:
- Use keyword-only arguments (after
*) for optional parameters with defaults - Boolean parameters SHOULD use keyword-only to avoid ambiguity
- Trailing comma after last parameter in multi-line signatures
- Opening paren stays with function name
Use Google-style docstrings:
def process_request(
request: Request,
*,
validate: bool = True,
) -> Response:
"""Process an incoming request and return the response.
Args:
request: The request to process.
validate: Whether to validate the request before processing.
Returns:
The processed response.
Raises:
ValidationError: If validation is enabled and the request is invalid.
ConnectionError: If the connection to the backend fails.
"""Rules:
- First line is a concise summary ending with a period
- Blank line between summary and sections
- Document all parameters in
Args:section - Document return value in
Returns:section (skip forNone) - Document exceptions in
Raises:section
Inline comments - use NOTE format:
# NOTE(username): Explanation of why this code exists.
result = do_something()Rules:
- Two spaces before inline comment
- Blank line ABOVE standalone comment (unless first line in block)
- NO blank line between comment and code it describes
- Prefix:
# NOTE(username):where username is your GitHub username - Comment on the 'why', not the 'what'
try:
result = await client.get_resource(resource_id)
except SpecificError as e:
logger.error("Failed to get resource '%s': %s", resource_id, e)
raise
except Exception as e:
raise ConnectorError(
f"Unexpected error retrieving resource '{resource_id}'."
) from eRules:
- Exception variable name:
e(or more specific likehttp_error) - Always chain exceptions with
from eto preserve context - Wrap inserted values in single quotes in error messages
- End error messages with period
- All exceptions must have descriptive messages — never raise exceptions without context
DO:
raise ValueError(f"Parameter 'connection_id' cannot be empty.")
raise ConnectorError(f"Operation '{operation_id}' is not supported.")DO NOT:
raise ValueError() # No message
raise Exception("error") # Non-descriptive, too genericUse f-strings for simple formatting:
message = f"Processing request '{request_id}' for user '{user_id}'."Use logging-style formatting for log messages:
logger.info("Processing request '%s' for user '%s'.", request_id, user_id)- Class docstring
- Class-level constants
__init__method__enter__/__exit__or__aenter__/__aexit__(context managers)- Properties
- Public methods
- Private methods (
_method) - Magic methods (
__str__,__repr__, etc.)
- Module docstring
from __future__ import annotations- Standard library imports
- Third-party imports
- Local imports
- Constants
- Type aliases
- Classes
- Functions
if __name__ == "__main__":block (if applicable)
| Anti-Pattern | Correct Pattern |
|---|---|
requests in async code |
aiohttp with async/await |
| Mutable default arguments | None with default inside function |
Bare except: |
except Exception as e: |
isinstance(x, str) for type unions |
Use typing.Union or | with type guards |
| Magic numbers | Named constants (e.g., DEFAULT_TIMEOUT_SECONDS) |
Magic strings (e.g., "type", "object") |
Named constants or enums |
list[0] without length check |
next(iter(list), None) or explicit validation |
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_get_user_returns_user_when_found():
"""Test that get_user returns the user when found."""
# Arrange
mock_client = AsyncMock()
mock_client.get.return_value = {"id": "123", "name": "Test User"}
# Act
result = await get_user(mock_client, user_id="123")
# Assert
assert result is not None
assert result.id == "123"Rules:
- Test function naming:
test_<method>_<scenario>_<expected>ortest_<behavior> - Use
pytest.mark.asynciofor async tests - Use
AsyncMockfor mocking async functions - Follow Arrange-Act-Assert pattern
- One assertion concept per test (multiple related asserts are OK)
- Branch naming:
feature/description,fix/description,docs/description - Never push directly to main
- Always create PR for review
The official release flow is the Azure DevOps eng/ci/library-release.yml
pipeline. It creates a release/{version} branch, updates
src/azure/connectors/__init__.py, builds the wheel and source distribution,
creates a GitHub draft release with the artifacts, uploads partner drops, and
then requires a manual partner-release/PyPI step. The package version is loaded
from azure.connectors.__version__ via src/pyproject.toml.
Run the ADO release pipeline with the intended version. The pipeline creates a
bare numeric Git tag such as 1.2.3 or 1.2.3b1; Python release tags are not
v-prefixed.
# Example versions accepted by the release pipeline:
1.2.3
1.2.3a1
1.2.3b1
1.2.3rc1Use Python/PEP 440 pre-release suffixes:
1.2.3a1 # Alpha
1.2.3b1 # Beta
1.2.3rc1 # Release candidateAfter artifacts are uploaded to the Azure SDK partner drops location, manually trigger the partner-release pipeline using the BlobPath printed by the official pipeline. Return to the original run and approve the final manual gate once the partner-release pipeline has started.
Do not delete or recreate a published release tag as a normal retry path. Release
tags are part of the supply-chain integrity boundary and are protected by tag
rulesets. If a release fails after a tag was pushed, use a new version. For a
retry of the same base package, use a PEP 440 post-release such as
1.2.3.post1.
# Run the official release pipeline again with a new version:
1.2.3.post1Note: PyPI does not allow re-uploading the same version. If the package was pushed successfully but the release failed, always use a new version number. Deleting or retagging an existing remote tag should be treated as break-glass admin work, not the standard process.
- Creates and validates a
release/{version}branch - Builds wheel and sdist with
python -m build - Creates a GitHub draft release with the distribution files attached
- Uploads distribution files to Azure SDK partner drops
- Guides the maintainer through the manual partner-release/PyPI publication
When adding a new connector client to the SDK, use the add-connector skill (.github/skills/add-connector/SKILL.md) for detailed templates and step-by-step guidance.
-
Create the connector module in
src/azure/connectors/:- File name should match the connector API name (e.g.,
office365.py,sharepointonline.py)
- File name should match the connector API name (e.g.,
-
Implement the client class following existing patterns:
from azure.connectors.sdk import ConnectorClient class Office365Client(ConnectorClient): """Client for Office 365 connector operations.""" async def send_email(self, ...) -> EmailResponse: ...
-
Export from
__init__.py— add the new client tosrc/azure/connectors/__init__.py -
Update
README.md— add the connector to the "Validated Connectors" table with status (✅ E2E Validatedor🔄 SDK Generated) and test count -
Add unit tests — create
test_<connector>.pyintests/following the pattern of existing tests:- Constructor tests
- Method tests with mocked responses
- Error handling tests
- Type serialization round-trips
-
Create sample usage file — create
sample_connector_usage_<connector>.pyinsamples/sample_connector_usage/following the pattern of existing samples (port from .NET if available) -
Update
samples/sample_connector_usage/README.md— add the new sample to the samples table -
Update
CHANGELOG.md— add the new connector to the[Unreleased]section under### Added -
Update the connection setup skill — add the connector's API name to the supported list in
.github/skills/connection-setup/SKILL.md(Step 2) -
Run all tests —
pytestmust pass with zero failures before committing -
Create a PR — reference the GitHub issue (e.g.,
Closes #9)
- Module follows naming conventions (
snake_case.py) - Client class has proper type hints on all public methods
- Docstrings present on class and all public methods
- Existing connector tests still pass (no regressions)
- New connector tests cover: constructor, mocked success, mocked error, exception handling
- Module exported in
__init__.py - README.md connector table updated
- Sample file created and compiles without errors
- Samples README updated
- CHANGELOG.md updated
- Connection setup skill updated