Skip to content

Use:feat(python-sdk): add core, context, and public sdk surface #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
usmanabbas7 merged 1 commit into from
Mar 30, 2026
Merged

Use:feat(python-sdk): add core, context, and public sdk surface #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion src/convertcom_sdk/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,31 @@
from .api.api_manager import ApiManager
from .config import DEFAULT_CONFIG, build_config
from .context import Context
from .core import Core
from .data.data_manager import DataManager
from .data.data_store_manager import DataStoreManager
from .events.event_manager import EventManager
from .experience.experience_manager import ExperienceManager
from .features.feature_manager import FeatureManager
from .segments.segments_manager import SegmentsManager
from .bucketing.bucketing_manager import BucketingAllocation, BucketingManager
from .enums import BucketingError, FeatureStatus, RuleError, SystemEvents
from .enums import BucketingError, EntityType, FeatureStatus, RuleError, SystemEvents
from .rules.rule_manager import RuleManager
from .sdk import ConvertSDK

__all__ = [
"ApiManager",
"BucketingAllocation",
"BucketingError",
"BucketingManager",
"build_config",
"Context",
"ConvertSDK",
"Core",
"DataManager",
"DataStoreManager",
"DEFAULT_CONFIG",
"EntityType",
"EventManager",
"ExperienceManager",
"FeatureManager",
Expand Down
25 changes: 25 additions & 0 deletions src/convertcom_sdk/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,25 @@
from __future__ import annotations

from collections.abc import Mapping
from typing import Any

from convertcom_sdk.utils.object_utils import object_deep_merge

DEFAULT_CONFIG: dict[str, Any] = {
"api": {"endpoint": {"config": "", "track": ""}},
"environment": "staging",
"bucketing": {"max_traffic": 10000, "hash_seed": 9999},
"data": {},
"dataStore": None,
"dataRefreshInterval": 300000,
"events": {"batch_size": 10, "release_interval": 1000},
"logger": {"logLevel": "debug", "customLoggers": []},
"rules": {"keys_case_sensitive": True, "comparisonProcessor": None},
"network": {"tracking": True, "cacheLevel": "default", "source": "python-sdk"},
"sdkKey": "",
"sdkKeySecret": "",
}


def build_config(config: Mapping[str, Any] | None = None) -> dict[str, Any]:
return object_deep_merge(DEFAULT_CONFIG, dict(config or {}))
296 changes: 296 additions & 0 deletions src/convertcom_sdk/context.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,296 @@
from __future__ import annotations

from collections.abc import Mapping
from typing import Any

from convertcom_sdk.enums import EntityType, SystemEvents
from convertcom_sdk.utils.object_utils import object_deep_merge, object_not_empty

ENTITY_TYPE_MAP = {
EntityType.AUDIENCE.value: "audiences",
EntityType.LOCATION.value: "locations",
EntityType.SEGMENT.value: "segments",
EntityType.FEATURE.value: "features",
EntityType.GOAL.value: "goals",
EntityType.EXPERIENCE.value: "experiences",
}


def _normalize_entity_type(entity_type: EntityType | str) -> EntityType:
if isinstance(entity_type, EntityType):
return entity_type
return EntityType(entity_type)


class Context:
def __init__(
self,
config: Mapping[str, Any] | None,
visitor_id: str | None,
*,
event_manager: Any,
experience_manager: Any,
feature_manager: Any,
segments_manager: Any,
data_manager: Any,
api_manager: Any,
visitor_properties: Mapping[str, Any] | None = None,
) -> None:
self._config = dict(config or {})
self._visitor_id = visitor_id
self._event_manager = event_manager
self._experience_manager = experience_manager
self._feature_manager = feature_manager
self._segments_manager = segments_manager
self._data_manager = data_manager
self._api_manager = api_manager
self._environment = self._config.get("environment")
self._visitor_properties: dict[str, Any] = {}
if object_not_empty(visitor_properties):
filtered = self._data_manager.filter_report_segments(visitor_properties)
if filtered["properties"]:
self._visitor_properties = dict(filtered["properties"])
self._segments_manager.put_segments(visitor_id, visitor_properties)

def _has_visitor(self) -> bool:
return bool(self._visitor_id)

def _base_attributes(
self,
attributes: Mapping[str, Any] | None = None,
*,
type_casting_default: bool | None = None,
) -> dict[str, Any]:
attributes = dict(attributes or {})
result = {
"visitorProperties": self.get_visitor_properties(attributes.get("visitorProperties")),
"locationProperties": attributes.get("locationProperties"),
"updateVisitorProperties": attributes.get("updateVisitorProperties"),
"environment": attributes.get("environment") or self._environment,
}
if type_casting_default is not None:
result["typeCasting"] = attributes.get("typeCasting", type_casting_default)
if "forceVariationId" in attributes:
result["forceVariationId"] = attributes.get("forceVariationId")
return result

def run_experience(self, experience_key: str, attributes: Mapping[str, Any] | None = None) -> Any:
if not self._has_visitor():
return None
bucketed_variation = self._experience_manager.select_variation(
self._visitor_id,
experience_key,
self._base_attributes(attributes),
)
if isinstance(bucketed_variation, Mapping):
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": experience_key,
"variationKey": bucketed_variation.get("key"),
},
None,
True,
)
return bucketed_variation

def runExperience(self, experience_key: str, attributes: Mapping[str, Any] | None = None) -> Any:
return self.run_experience(experience_key, attributes)

def run_experiences(self, attributes: Mapping[str, Any] | None = None) -> list[Any] | None:
if not self._has_visitor():
return None
bucketed_variations = self._experience_manager.select_variations(
self._visitor_id,
self._base_attributes(attributes),
)
for bucketed_variation in bucketed_variations:
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": bucketed_variation.get("experienceKey"),
"variationKey": bucketed_variation.get("key"),
},
None,
True,
)
return bucketed_variations

def runExperiences(self, attributes: Mapping[str, Any] | None = None) -> list[Any] | None:
return self.run_experiences(attributes)

def run_feature(self, key: str, attributes: Mapping[str, Any] | None = None) -> Any:
if not self._has_visitor():
return None
attributes = dict(attributes or {})
bucketed_feature = self._feature_manager.run_feature(
self._visitor_id,
key,
self._base_attributes(attributes, type_casting_default=True),
attributes.get("experienceKeys"),
)
items = bucketed_feature if isinstance(bucketed_feature, list) else [bucketed_feature]
for item in [feature for feature in items if isinstance(feature, Mapping)]:
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": item.get("experienceKey"),
"featureKey": key,
"status": item.get("status"),
},
None,
True,
)
Comment on lines +135 to +146

@gemini-code-assist gemini-code-assist Bot Mar 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The list comprehension [feature for feature in items if isinstance(feature, Mapping)] creates a new temporary list just for iteration. This is slightly inefficient and can be made more readable by using a simple for loop with a check inside.

Suggested change
for item in [feature for feature in items if isinstance(feature, Mapping)]:
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": item.get("experienceKey"),
"featureKey": key,
"status": item.get("status"),
},
None,
True,
)
for item in items:
if isinstance(item, Mapping):
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": item.get("experienceKey"),
"featureKey": key,
"status": item.get("status"),
},
None,
True,
)

return bucketed_feature

def runFeature(self, key: str, attributes: Mapping[str, Any] | None = None) -> Any:
return self.run_feature(key, attributes)

def run_features(self, attributes: Mapping[str, Any] | None = None) -> list[Any] | None:
if not self._has_visitor():
return None
bucketed_features = self._feature_manager.run_features(
self._visitor_id,
self._base_attributes(attributes, type_casting_default=True),
)
for item in [feature for feature in bucketed_features if isinstance(feature, Mapping)]:
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": item.get("experienceKey"),
"featureKey": item.get("key"),
"status": item.get("status"),
},
None,
True,
)
Comment on lines +159 to +170

@gemini-code-assist gemini-code-assist Bot Mar 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similar to run_feature, the list comprehension here creates an unnecessary temporary list. A simple for loop with a check would be more efficient and readable.

Suggested change
for item in [feature for feature in bucketed_features if isinstance(feature, Mapping)]:
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": item.get("experienceKey"),
"featureKey": item.get("key"),
"status": item.get("status"),
},
None,
True,
)
for item in bucketed_features:
if isinstance(item, Mapping):
self._event_manager.fire(
SystemEvents.BUCKETING,
{
"visitorId": self._visitor_id,
"experienceKey": item.get("experienceKey"),
"featureKey": item.get("key"),
"status": item.get("status"),
},
None,
True,
)

return bucketed_features

def runFeatures(self, attributes: Mapping[str, Any] | None = None) -> list[Any] | None:
return self.run_features(attributes)

def track_conversion(self, goal_key: str, attributes: Mapping[str, Any] | None = None) -> Any:
if not self._has_visitor():
return None
attributes = dict(attributes or {})
goal_data = attributes.get("conversionData")
if goal_data is not None and not isinstance(goal_data, list):
return None
result = self._data_manager.convert(
self._visitor_id,
goal_key,
attributes.get("ruleData"),
goal_data,
self._segments_manager.get_segments(self._visitor_id),
attributes.get("conversionSetting"),
)
if result:
self._event_manager.fire(
SystemEvents.CONVERSION,
{"visitorId": self._visitor_id, "goalKey": goal_key},
None,
True,
)
return result

def trackConversion(self, goal_key: str, attributes: Mapping[str, Any] | None = None) -> Any:
return self.track_conversion(goal_key, attributes)

def set_default_segments(self, segments: Mapping[str, Any]) -> None:
self._segments_manager.put_segments(self._visitor_id, segments)

def setDefaultSegments(self, segments: Mapping[str, Any]) -> None:
self.set_default_segments(segments)

def set_custom_segments(self, segment_keys: list[str] | str, attributes: Mapping[str, Any] | None = None) -> Any:
return self.run_custom_segments(segment_keys, attributes)

def setCustomSegments(self, segment_keys: list[str] | str, attributes: Mapping[str, Any] | None = None) -> Any:
return self.set_custom_segments(segment_keys, attributes)

def run_custom_segments(self, segment_keys: list[str] | str, attributes: Mapping[str, Any] | None = None) -> Any:
if not self._has_visitor():
return None
keys = [segment_keys] if isinstance(segment_keys, str) else list(segment_keys)
return self._segments_manager.select_custom_segments(
self._visitor_id,
keys,
self.get_visitor_properties((attributes or {}).get("ruleData")),
)

def runCustomSegments(self, segment_keys: list[str] | str, attributes: Mapping[str, Any] | None = None) -> Any:
return self.run_custom_segments(segment_keys, attributes)

def update_visitor_properties(self, visitor_id: str, visitor_properties: Mapping[str, Any]) -> None:
self._data_manager.put_data(visitor_id, {"segments": dict(visitor_properties)})
Comment on lines +228 to +229

@gemini-code-assist gemini-code-assist Bot Mar 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The update_visitor_properties method on the Context class accepts a visitor_id. This is confusing and potentially error-prone, as a Context instance is already associated with a specific visitor (self._visitor_id). This design could lead to unintentionally modifying data for a different visitor from within a specific visitor's context.

To improve clarity and prevent misuse, this method should operate on the visitor associated with the current context. I recommend removing the visitor_id parameter and using self._visitor_id instead. The aliased method updateVisitorProperties should also be updated to reflect this change in signature.

Suggested change
def update_visitor_properties(self, visitor_id: str, visitor_properties: Mapping[str, Any]) -> None:
self._data_manager.put_data(visitor_id, {"segments": dict(visitor_properties)})
def update_visitor_properties(self, visitor_properties: Mapping[str, Any]) -> None:
self._data_manager.put_data(self._visitor_id, {"segments": dict(visitor_properties)})


def updateVisitorProperties(self, visitor_id: str, visitor_properties: Mapping[str, Any]) -> None:
self.update_visitor_properties(visitor_id, visitor_properties)

def get_config_entity(self, key: str, entity_type: EntityType | str) -> Any:
entity_type = _normalize_entity_type(entity_type)
if entity_type == EntityType.VARIATION:
for experience in self._data_manager.get_entities_list("experiences"):
variation = self._data_manager.get_sub_item(
"experiences",
experience.get("key"),
"variations",
key,
"key",
"key",
)
if variation:
return variation
return None
Comment on lines +236 to +248

@gemini-code-assist gemini-code-assist Bot Mar 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

When entity_type is EntityType.VARIATION, this method iterates through all experiences to find a matching variation. This can be inefficient if there are many experiences, leading to O(N) complexity where N is the number of experiences.

For better performance, consider optimizing the data structure within DataManager to allow for a more direct lookup of variations, for example by pre-processing the config data to create a map of variation keys/IDs to their parent experience.

return self._data_manager.get_entity(key, ENTITY_TYPE_MAP[entity_type.value])

def getConfigEntity(self, key: str, entity_type: EntityType | str) -> Any:
return self.get_config_entity(key, entity_type)

def get_config_entity_by_id(self, entity_id: str, entity_type: EntityType | str) -> Any:
entity_type = _normalize_entity_type(entity_type)
if entity_type == EntityType.VARIATION:
for experience in self._data_manager.get_entities_list("experiences"):
variation = self._data_manager.get_sub_item(
"experiences",
experience.get("id"),
"variations",
entity_id,
"id",
"id",
)
if variation:
return variation
return None
return self._data_manager.get_entity_by_id(entity_id, ENTITY_TYPE_MAP[entity_type.value])

def getConfigEntityById(self, entity_id: str, entity_type: EntityType | str) -> Any:
return self.get_config_entity_by_id(entity_id, entity_type)

def get_visitor_data(self) -> dict[str, Any]:
return self._data_manager.get_data(self._visitor_id) or {}

def getVisitorData(self) -> dict[str, Any]:
return self.get_visitor_data()

def release_queues(self, reason: str | None = None) -> Any:
if self._data_manager.data_store_manager:
self._data_manager.data_store_manager.release_queue(reason)
return self._api_manager.release_queue(reason)

def releaseQueues(self, reason: str | None = None) -> Any:
return self.release_queues(reason)

def get_visitor_properties(self, attributes: Mapping[str, Any] | None = None) -> dict[str, Any]:
store_data = self._data_manager.get_data(self._visitor_id) or {}
segments = store_data.get("segments") or {}
visitor_properties = (
object_deep_merge(self._visitor_properties, dict(attributes or {}))
if attributes
else dict(self._visitor_properties)
)
return object_deep_merge(segments, visitor_properties)
Loading