NOTICE: All information contained herein is, and remains
the property of TechnoCore Automate.
ObjChannelmail provides classes for managing and sending emails through defined channels, handling mailing lists, and processing email stages.
ChannelMail(ObjSupervisor.Supervisor)Manages individual email entries within a channel, including reading, creating, updating, and marking email status.
__init__(self, db: Objects.DatabaseConnection = 0)Initializes the ChannelMail object.
db: An optional database connection object.Read(self, channel: str, email: str)Reads an email entry for a specific channel and email address from the database.
channel: The name of the channel.email: The email address to read.Example:
channel_mail = ChannelMail(db_connection)
channel_mail.Read("newsletter", "test@example.com")
ReadOnGuid(self, uuid)Reads an email entry based on its UUID.
uuid: The UUID of the email entry.Example:
channel_mail = ChannelMail(db_connection)
channel_mail.ReadOnGuid("some-uuid-string")
Validate(self, uuid: str = "")Validates an email entry by setting its _Active status to 1 and updating it in the database.
uuid: The UUID of the email entry to validate.Example:
channel_mail = ChannelMail(db_connection)
status = channel_mail.Validate("some-uuid-string")
Create(self, channel: str, email: str = "")Creates a new email entry in the def_ChannelMail table.
channel: The channel name for the new entry.email: The email address for the new entry.Example:
channel_mail = ChannelMail(db_connection)
sql_query = channel_mail.Create("newsletter", "new@example.com")
Update(self)Updates the current email entry in the database.
Example:
channel_mail = ChannelMail(db_connection)
channel_mail.Read("newsletter", "test@example.com")
channel_mail._Active = 0
sql_query = channel_mail.Update()
SetGuid(self)Sets a new UUID for the email entry and marks _Clickthrough as "N", then updates the entry.
Example:
channel_mail = ChannelMail(db_connection)
channel_mail.Read("newsletter", "test@example.com")
channel_mail.SetGuid()
mark_visit(self)Marks an email as visited (_Clickthrough = "Y") and updates the Clickdate.
Example:
channel_mail = ChannelMail(db_connection)
channel_mail.Read("newsletter", "test@example.com")
channel_mail.mark_visit()
mark_mail(self, message="", description="", notes="", footer="", item_id=0)Marks an email as sent (_SendStatus = "MAILED") and updates the Senddate.
message: The message content.description: The description of the email.notes: Additional notes.footer: The email footer.item_id: The ID of the item in the stage table.Example:
channel_mail = ChannelMail(db_connection)
channel_mail.Read("newsletter", "test@example.com")
channel_mail.mark_mail("Email content", "Subject", "Some notes")
ChannelMailSet(ObjSupervisor.Supervisor)Manages a set of ChannelMail objects, typically for mailing list operations.
__init__(self, db: Objects.DatabaseConnection = 0)Initializes the ChannelMailSet object.
db: An optional database connection object.Read(self, channel: str, age: int = 0)Reads a list of active email addresses for a given channel, optionally filtered by age.
channel: The channel name.age: Minimum age in days for emails to be included.Example:
channel_mail_set = ChannelMailSet(db_connection)
channel_mail_set.Read("newsletter", age=7)
SetGuid(self)Assigns a new UUID to each email in the item_list.
Example:
channel_mail_set = ChannelMailSet(db_connection)
channel_mail_set.Read("newsletter")
channel_mail_set.SetGuid()
users_to_channel(self, channel)Inserts users from the User table into the def_Channelmail table for a given channel.
channel: The target channel name.Example:
channel_mail_set = ChannelMailSet(db_connection)
channel_mail_set.users_to_channel("new_channel")
Mail(self, check="proof")Sends emails to the mailing list based on the check parameter ("proof" or "release").
check: "proof" to send proof emails, "release" to send actual emails.Example:
channel_mail_set = ChannelMailSet(db_connection)
channel_mail_set.Read("newsletter")
status = channel_mail_set.Mail("release")
Install(self)Installs the necessary database objects for channel mail.
Example:
channel_mail_set = ChannelMailSet(db_connection)
channel_mail_set.Install()
ChannelDirectSet(ChannelMailSet)Extends ChannelMailSet to handle direct email sending based on staged channel data.
__init__(self, db: Objects.DatabaseConnection = 0)Initializes the ChannelDirectSet object.
db: An optional database connection object.Read(self, channel="", context=[])Reads channel data directly from the stage_Channel table.
channel: The channel name.context: A list of context information.Example:
channel_direct_set = ChannelDirectSet(db_connection)
channel_direct_set.Read("direct_channel")
channel_sender(object)A helper class for sending individual emails within a separate thread or process.
__init__(self, guid: str, channel_obj: ObjChannel.ObjChannel, email: str)Initializes the channel_sender object.
guid: The UUID of the email.channel_obj: An instance of ObjChannel.ObjChannel.email: The recipient's email address.run(self)Executes the email sending process for a single email.
ChannelStage(ObjData.ObjData)Manages the staging of channel emails, including rendering and processing.
__init__(self, db: Objects.DatabaseConnection = 0, item_id: str = "")Initializes the ChannelStage object.
db: An optional database connection object.item_id: An optional item ID.patch_param(self, alt: str, depth=0)Replaces placeholders in a string with corresponding values from the object's attributes or context.
alt: The string with placeholders to be patched.depth: Recursion depth (for internal use).Update(self)Updates the current stage entry in the database.
Read(self, guid: str, context=[])Reads a channel stage entry from the database based on GUID or context.
guid: The GUID of the stage entry.context: A list or dictionary of context information.Render(self, context=dict())Renders and processes the email for a channel stage, handling mailing lists or direct sends.
context: A dictionary of context information.ChannelStageSet(ObjSupervisor.Supervisor)Manages a set of ChannelStage objects, typically for processing multiple staged emails.
__init__(self, db: Objects.DatabaseConnection = 0, item_id: str = "")Initializes the ChannelStageSet object.
db: An optional database connection object.item_id: An optional item ID.Create(self)Builds the set of channel stages.
Read(self, guid: str = "")Reads and processes a channel stage based on its GUID.
guid: The GUID of the stage to read.Build(self)Placeholder for building the set of channel stages.
ActionResponses(self)Processes action responses, such as updating click-through tracking.
launch_service(self, thread_count: str = "worker")Launches the channel mail service.
thread_count: The type of service to launch ("worker" or "dispatcher").micro_service(self, role="")Runs the channel mail microservice.
role: The role of the microservice (e.g., "dispatcher", "worker").run_direct(self, context: dict = dict(), channel: str = "", guid: str = "")Runs a direct channel mail operation.
context: A dictionary of context information.channel: The channel name.guid: The GUID for the operation.The module also provides a command-line interface using typer for various operations:
stage(guid: str)Processes a channel stage.
guid: The GUID of the stage to process.micro_service(role: str = "dispatcher")Runs the channel mail microservice.
role: The role of the microservice (e.g., "dispatcher", "worker").dispatcher()Runs the channel mail microservice dispatcher.
worker()Runs the channel mail microservice worker.
default(channel: str)Default command to run a direct channel test.
channel: The channel name for the test.The ObjChannelmail module has comprehensive test coverage for email channel functionality.
| Test Suite | Tests | Type | Purpose |
|---|---|---|---|
test_ObjChannelmail.py |
73 | Unit | Email composition, SMTP connection, template processing, attachment handling |
| Total | 73 (71 passed, 2 skipped) |
# Run all ObjChannelmail tests
pytest resource.test/pytests/factory.core/test_ObjChannelmail.py -v