This document outlines the primary coding style rules and conventions to be followed for this project.
All local variables and function/method parameters should use snake_case. Names should be descriptive and meaningful.
worker_queue, file_path, data_import_codeq, fp, dicAll function and method names should use snake_case, including CRUD operations.
read(), create(), update(), delete(), get_tasks(), add_comment()Read(), Create(), Update(), Delete(), GetTasks(), AddComment()Module-level constants should use ALL_CAPS with underscores:
DEF_VERSION = "8.0"
MAX_RETRY_COUNT = 3
DEFAULT_TIMEOUT = 30
BATCH_SIZE = 100
Boolean variables and parameters should use is_, has_, or can_ prefixes:
is_valid, has_permission, can_edit, is_activevalid, permission, edit, activePrivate member variables have a specific and strict naming convention:
_).Examples:
self._data_import_code becomes self._Dataimportcodeself._column_types becomes self._Columntypesself._file_guid becomes self._Fileguidself._table_name becomes self._TablenameDefine enumerations centrally in factory.core/ObjEnum.py for reuse across modules:
# In ObjEnum.py
from enum import StrEnum
class TicketStatus(StrEnum):
NEW = "NEW"
TRIAGED = "TRIAGED"
IN_PROGRESS = "IN_PROGRESS"
RESOLVED = "RESOLVED"
CLOSED = "CLOSED"
class TicketPriority(StrEnum):
CRITICAL = "CRITICAL"
HIGH = "HIGH"
MEDIUM = "MEDIUM"
LOW = "LOW"
Usage in other modules:
from ObjEnum import TicketStatus, TicketPriority
if ticket._Status == TicketStatus.NEW:
...
Always use f-strings for string formatting and concatenation.
self.debug(f"Processing {file_path} with code {data_import_code}")self.debug("Processing " + file_path + " with code " + data_import_code)Avoid naked except blocks. Always catch a specific exception, and if you must catch a broad exception, log the error.
except OSError as e: self.debug(f"Could not move file: {e}")except: passThe maximum line length for Python code is 80 characters.
Run ruff check on any Python file that is modified to ensure it adheres to project linting standards.
Organize class methods in the following order:
__init__ and other dunder methods (__str__, __repr__, etc.)read, create, update, delete)_method_name)class ObjTicket(ObjData.ObjData):
def __init__(self, DB: int = 0) -> None:
super().__init__(DB)
# initialization...
def __str__(self) -> str:
return f"Ticket #{self._Ticketnumber}"
# CRUD methods
def read(self, guid: str) -> int:
...
def create(self, subject: str, ...) -> str:
...
def update(self) -> None:
...
# Public methods
def assign(self, user: str, action_by: str) -> None:
...
def resolve(self, action_by: str) -> None:
...
# Private methods
def _ensure_loaded(self) -> None:
...
def _build_ticket_context(self) -> dict:
...
Do not use unexplained numeric literals in code. Define named constants instead:
BAD:
if retry_count > 3:
raise Exception("Too many retries")
time.sleep(300)
GOOD:
MAX_RETRY_COUNT = 3
CACHE_TTL_SECONDS = 300
if retry_count > MAX_RETRY_COUNT:
raise Exception("Too many retries")
time.sleep(CACHE_TTL_SECONDS)
Exception: 0, 1, -1 are acceptable when their meaning is obvious from context.
Use self.debug() for all logging and debugging output. Do not use print() or other logging libraries.
If an if DO_DEBUG: block contains only self.debug() statements, the if DO_DEBUG: wrapper should be removed, leaving only the self.debug() calls.
All Python scripts that require a command-line interface must use the typer library.
"""...""") to enhance readability.self.escape_sql() to escape values being inserted into SQL queries.__init__)Do not use self._IsA, self._Version, or self._Updatedate within the __init__ method of objects.
Use the following pattern at the beginning of a script (after external package imports) to ensure local modules can be imported correctly:
base_path = os.getcwd()
paths = [
"",
"/factory.core",
"/factory.service",
# ... other necessary paths
]
for relative_path in paths:
if (base_path + relative_path) not in sys.path:
sys.path.append(base_path + relative_path)
All .md documentation files must begin with the following header:
# (c) Technocore - All Rights Reserved.
NOTICE: All information contained herein is, and remains
the property of TechnoCore.
The intellectual and technical concepts contained
herein are proprietary to TechnoCore and dissemination of this information or reproduction of this material
is strictly forbidden unless prior written permission is obtained
from TechnoCore.
### 3.8. YAML Files
Never include `---` document separators in YAML files. All YAML files should be single-document only.
### 3.9. Module Documentation
All Python `Obj*.py` files must have corresponding `.md` documentation in the same folder.
- Example: `factory.core/ObjNotify.py` → `factory.core/ObjNotify.md`
Update `.md` files when making significant changes to Python files.
### 3.10. Module Constants
Every module should have the following constants defined at the top:
```python
DEF_VERSION = "8.0"
# ... imports ...
DO_DEBUG: bool = False
The DO_DEBUG constant must always be present, even if unused within the module (it is used at runtime externally).
Use PEP 350 codetags with a trailing <>:
# TODO: Description <>
# FIXME: Description <>
# BUG: Description <>
# NOTE: Description <>
# HACK: Description <>
# SEE: Reference <>
Project-specific codetags:
# TABLE: table_name <>
# QUERY: query description <>
Use type hints for function parameters and return values:
def process_file(file_path: str, max_size: int = 100) -> bool:
...
Write brief docstrings that explain the purpose and logic of functions. Do not reiterate all parameters unless clarification is needed.
GOOD:
def calculate_sla_deadline(self, priority: str) -> datetime:
"""Calculate SLA deadline based on priority config."""
BAD:
def calculate_sla_deadline(self, priority: str) -> datetime:
"""
Calculate SLA deadline.
Args:
priority: The priority string.
Returns:
datetime: The deadline.
"""
Organize imports in the following order, separated by blank lines:
import os
import sys
from datetime import datetime
import typer
from fuzzywuzzy import fuzz
base_path = os.getcwd()
paths = [
"",
"/factory.core",
"/factory.service",
]
for relative_path in paths:
if (base_path + relative_path) not in sys.path:
sys.path.append(base_path + relative_path)
import ObjData
from ObjTicket import ObjTicket
Use Objects.global_config to access configuration from config.yaml:
import Objects
ini = Objects.global_config
# Get string value
host = ini.Get("database", "host", "localhost")
# Get integer value
port = ini.GetInt("database", "port", 3306)
# Get float value
timeout = ini.GetFloat("system", "timeout", 30.0)
# Access the current package
package = ini.package # e.g., "homechoice"
| Method | Purpose | Returns |
|---|---|---|
ini.Get(section, option, default="") |
Get string value | str |
ini.GetInt(section, option, default=0) |
Get integer value | int |
ini.GetFloat(section, option, default) |
Get float value | float |
Use self.get_package() within classes inheriting from ObjData:
class ObjTicket(ObjData.ObjData):
def create(self, subject: str) -> str:
package = self.get_package() # Returns current package
...
Use self.get_constants() to retrieve constants from the module's YAML file:
# In ObjTicket.yaml:
# constants:
# max_tasks: 50
# default_priority: "MEDIUM"
# In ObjTicket.py:
max_tasks = self.get_constants("max_tasks") # Returns 50
priority = self.get_constants("default_priority") # Returns "MEDIUM"
Always use self.escape_sql() for user-provided values. Never concatenate raw input.
| Scenario | Approach |
|---|---|
| User-provided values | Always escape_sql() |
| Column/table names | Use allowlists, never escape |
| Constrained values (status, priority) | Use enums from ObjEnum |
| LIKE clauses | Escape value AND escape %, _ wildcards |
| Numeric values | Cast to int/float before use |
# GOOD - escaped user input
sql = f"""
SELECT * FROM data_ticket
WHERE Guid = '{self.escape_sql(guid)}'
AND Status = '{self.escape_sql(status)}'
"""
# GOOD - enum for constrained values
from ObjEnum import TicketStatus
if status not in list(TicketStatus):
raise ValueError(f"Invalid status: {status}")
# GOOD - numeric casting
limit = int(user_limit)
sql = f"SELECT * FROM data_ticket LIMIT {limit}"
# GOOD - column allowlist
ALLOWED_COLUMNS = ["Subject", "Status", "Priority"]
if sort_column not in ALLOWED_COLUMNS:
raise ValueError(f"Invalid column: {sort_column}")
sql = f"SELECT * FROM data_ticket ORDER BY {sort_column}"
# BAD - raw concatenation
sql = f"SELECT * FROM data_ticket WHERE Guid = '{guid}'"
# BAD - dynamic column without validation
sql = f"SELECT * FROM data_ticket ORDER BY {user_column}"
Validate at system entry points (API, forms, CLI):
def create(self, subject: str, priority: str) -> str:
# Validate constrained values
if priority not in list(TicketPriority):
raise ValueError(f"Invalid priority: {priority}")
# Length limits
if len(subject) > 255:
raise ValueError("Subject too long")
# Then safe to use with escape_sql()
sql = self.get_queries("create_ticket").format(
subject=self.escape_sql(subject),
priority=self.escape_sql(priority),
)
self.get_queries("query_name") to retrieve SQL from YAML files.self.create_tables_from_yaml() to create tables from schema definitions..yaml file (e.g., ObjTicket.yaml).| Function | Purpose | Returns |
|---|---|---|
sql_execute(sql) |
Execute INSERT, UPDATE, DELETE statements | None |
sql_read_object(sql) |
Load a single row into object's _Property attributes |
Row count (0 or 1) |
sql_get_value(sql) |
Get a single scalar value | Value or None |
sql_get_dictionary(sql) |
Get a single row as a dictionary | Dict or None |
sql_get_dictionary_list(sql) |
Get multiple rows as list of dictionaries | List[Dict] |
sql_get_list(sql) |
Get a single column as a list | List |
Always use self.escape_sql() to escape values in SQL queries:
sql = self.get_queries("read_ticket").format(
guid=self.escape_sql(guid),
status=self.escape_sql(status),
)
result = self.sql_read_object(sql)
Note: Use escape_sql(), not sql_escape() (deprecated).
# Execute an INSERT
sql = self.get_queries("create_ticket").format(
guid=guid,
subject=self.escape_sql(subject),
)
self.sql_execute(sql)
# Read into object properties
sql = self.get_queries("read_ticket").format(
guid=self.escape_sql(guid),
)
if self.sql_read_object(sql):
print(self._Subject) # Property populated from row
# Get a single value
sql = self.get_queries("count_open").format(
status=self.escape_sql("NEW"),
)
count = self.sql_get_value(sql)
# Get list of dictionaries
sql = self.get_queries("list_tickets").format(
status=self.escape_sql("NEW"),
)
tickets = self.sql_get_dictionary_list(sql)
for ticket in tickets:
print(ticket["Subject"])
Test files belong in resource.test/pytests/ mirroring the factory structure:
factory.core tests → resource.test/pytests/factory.core/factory.service tests → resource.test/pytests/factory.service/factory.text/package.web tests → resource.test/pytests/factory.text/package.web/Test file naming: test_<ModuleName>.py (e.g., test_ObjTicket.py)
The following folders must always be in .gitignore:
data.configlocal.documentsdata.documentsarchive.documentsVerify these are present before every commit.
DEPRECATED: to the docstring with removal versionself.debug()def old_method(self, arg: str) -> str:
"""
DEPRECATED: This method will be removed in v9.0.
Use new_method() instead.
"""
def old_method(self, arg: str) -> str:
"""
DEPRECATED: This method will be removed in v9.0.
Use new_method() instead.
"""
self.debug(
":warning: WARNING: old_method is deprecated, "
"use new_method instead"
)
return self.new_method(arg)
if ini.Get("redis", "redisip"):
self.debug(
"WARNING: 'redis.redisip' config is deprecated. "
"Use 'redis.host' instead."
)
Prefer early returns to reduce nesting and improve readability:
BAD:
def process_ticket(self, guid: str) -> bool:
if guid:
result = self.read(guid)
if result:
if self._Status != "CLOSED":
# actual logic here
return True
else:
return False
else:
return False
else:
return False
GOOD:
def process_ticket(self, guid: str) -> bool:
if not guid:
return False
result = self.read(guid)
if not result:
return False
if self._Status == "CLOSED":
return False
# actual logic here
return True
Import from factory.core/Decorators.py:
import Decorators
| Decorator | Purpose | Parameters |
|---|---|---|
@Decorators.transient_lru_cache |
Time-based LRU cache that auto-expires | seconds=600, maxsize=128 |
@Decorators.retry |
Retry function on failure | max_attempts, delay=1 |
import Decorators
@Decorators.transient_lru_cache(seconds=3600, maxsize=128)
def get_user_permissions(self, user: str) -> list:
"""Cache user permissions for 1 hour."""
sql = self.get_queries("get_permissions").format(
user=self.escape_sql(user)
)
return self.sql_get_dictionary_list(sql)
@Decorators.retry(max_attempts=3, delay=2)
def fetch_external_api(self, url: str) -> dict:
"""Retry API call up to 3 times with 2s delay."""
response = requests.get(url)
return response.json()
Use standard decorators where appropriate:
@staticmethod
def format_date(value: str) -> str:
"""No access to self needed."""
...
@classmethod
def from_dict(cls, data: dict) -> "ObjTicket":
"""Alternative constructor."""
...
@property
def is_overdue(self) -> bool:
"""Computed property."""
return datetime.now() > self._Deadline
Use with statements for resource management:
# File operations
with open(file_path, "r") as f:
content = f.read()
# Database cursors (when not using ObjData methods)
with self.get_db_cursor(db) as cursor:
cursor.execute(sql)
result = cursor.fetchall()
# Locks
with self._lock:
self._shared_resource += 1
Catch specific exceptions, not generic Exception:
# GOOD
try:
result = int(value)
except ValueError as e:
self.debug(f"Invalid integer value: {e}")
result = 0
# BAD
try:
result = int(value)
except Exception:
result = 0
When catching and re-raising, add context:
try:
self.sql_execute(sql)
except DatabaseError as e:
self.debug(f"Failed to update ticket {self._Guid}: {e}")
raise
# Expected error - handle gracefully
try:
result = self.read(guid)
except RecordNotFoundError:
return None
# Unexpected error - log and re-raise
try:
self._process_complex_operation()
except Exception as e:
self.debug(f"Unexpected error in process: {e}")
raise
Ensure tests don't pollute global state:
@pytest.fixture
def ticket_instance():
"""Create isolated test instance."""
original_config = Objects.global_config
Objects.global_config = MockConfigIni()
instance = ObjTicket()
yield instance
# Restore original state
Objects.global_config = original_config
from unittest import mock
def test_send_notification(ticket_instance):
with mock.patch.object(
ticket_instance, "sql_execute"
) as mock_sql:
mock_sql.return_value = None
ticket_instance.create(subject="Test")
mock_sql.assert_called()
Use descriptive test names:
# GOOD
def test_create_ticket_with_valid_subject_returns_guid():
...
def test_read_ticket_with_invalid_guid_returns_zero():
...
# BAD
def test_create():
...
def test_1():
...
Every module with CLI should follow this pattern:
import typer
# ... class definition ...
app = typer.Typer()
@app.command()
def cli_read(guid: str) -> None:
"""Read a record by guid."""
obj = ObjTicket()
result = obj.read(guid)
if result:
obj.debug(f"Found: {obj._Subject}")
else:
obj.debug("Not found")
@app.command()
def cli_create(
subject: str,
priority: str = "MEDIUM",
) -> None:
"""Create a new record."""
obj = ObjTicket()
guid = obj.create(subject=subject, priority=priority)
obj.debug(f"Created: {guid}")
@app.command()
def list_open(package: str = "") -> None:
"""List open records."""
obj = ObjTicket()
for item in obj.get_open_tickets(package):
obj.debug(f"{item.get('Subject')}")
if __name__ == "__main__":
app()
cli_ when the command name conflicts with a class methodUse ObjNotify for sending notifications:
import ObjNotify
def notify_event(
self,
notify_code: str,
context: dict = None,
) -> None:
"""Send notification using configured channel."""
if context is None:
context = {}
context["TicketNumber"] = self._Ticketnumber
context["Subject"] = self._Subject
context["NotifyCode"] = notify_code
notifier = ObjNotify.ObjNotify(self.DB)
notifier.Read(notify_code)
notifier.Run(notify_code, context=context)
Notify codes are configured in the database and define:
For method results that can be cached with time-based expiry:
@Decorators.transient_lru_cache(seconds=600, maxsize=128)
def get_user_details(self, user: str) -> dict:
"""Cache user details for 10 minutes."""
sql = self.get_queries("get_user").format(
user=self.escape_sql(user)
)
return self.sql_get_dictionary(sql)
For shared cache across processes, use Redis via ObjData methods:
# Get cached value
value = self.redis_get(f"ticket:{guid}")
if value is None:
value = self._expensive_lookup(guid)
self.redis_set(f"ticket:{guid}", value, ttl=3600)
Clear caches when data changes:
def update(self) -> None:
"""Update record and invalidate cache."""
self.sql_execute(sql)
# Clear any cached data for this record
self.get_user_details.cache_clear() # For lru_cache
self.redis_delete(f"ticket:{self._Guid}") # For Redis
Factory modules contain the core business logic, organized by domain:
| Folder | Purpose | Example Files |
|---|---|---|
factory.core |
Fundamental shared components, base classes, utilities | ObjData.py, ObjWorkflow.py, ObjConnection.py |
factory.service |
Business service implementations, calculations, classifications | ObjServiceCalculation.py, ObjServiceClassification.py |
factory.report |
Report generation and data presentation | ObjReportBigTable.py, ObjReportChart.py |
factory.web |
Web server components, form handling | ObjWebForm.py, ObjWebMail.py |
factory.conversation |
Chat and messaging logic (Discord, WhatsApp, Console) | ObjConversation.py |
factory.ai |
AI/ML integrations (image generation, LLM, time-series) | ObjAIImage.py, ObjAILlm.py |
factory.webhook |
Webhook handlers | ObjWebhook.py |
factory.deploy |
Deployment, compilation, configuration tools | ObjCompile.py, ObjLxc.py |
factory.field |
Form field definitions | ObjField*.py |
factory.export |
Data export formats (CSV, JSON, YAML, Excel, Markdown) | ObjDataExportCsv.py, ObjDataExportJson.py |
factory.import |
Data import handlers | ObjDataImportCSV.py, ObjDataImportParquet.py |
factory.text |
Text processing utilities | ObjProcessText.py |
factory.sms |
SMS integration | ObjSms*.py |
factory.pages |
Page templates | ObjPage*.py |
factory.tui |
Terminal user interface components | ObjTui*.py |
factory.workflow |
Workflow-specific components | ObjWorkflow*.py |
factory.imap |
Email/IMAP integration | ObjImap.py |
factory.formlayout |
Form layout definitions | ObjFormlayout*.py |
factory.learn |
Machine learning components | ObjLearn*.py |
Client-specific implementations go in package.* subdirectories within factory modules:
factory.service/
├── package.core/ # Core/shared services
├── package.homechoice/ # HomeChoice client services
├── package.fullhouse/ # FullHouse client services
└── package.chrysalistp/ # Chrysalis client services
Resource folders contain non-code assets and configuration:
| Folder | Purpose | Example Files |
|---|---|---|
resource.config |
Configuration files, requirements, install scripts | requirements.txt, mariadb.yaml, mongodb.yaml |
resource.bin |
Shell scripts for Docker, backups, system setup | start_mongo.sh, backup_mysql.sh |
resource.templates |
Email/report HTML templates with YAML config | axion.html, axion.yaml |
resource.test |
Test data files, sample documents | Bank-Statement-Test1.pdf, MOCK_DATA.csv |
resource.test/pytests |
Pytest test suite (mirrors factory structure) | test_ObjData.py, test_ObjTicket.py |
resource.notes |
Documentation, how-to guides, policies | coding_style.md, howto/*.md |
resource.web |
JavaScript, CSS, web assets, fonts | javascript/, fonts/, icons/ |
resource.images |
Static images | logo.png, icons/ |
resource.docker |
Docker-related configuration | Dockerfile, docker-compose.yaml |
resource.pem |
SSL certificates and keys | *.pem, *.crt |
resource.setup |
Setup and installation scripts | install.sh |
resource.style |
CSS stylesheets | *.css |
resource.integrations |
Third-party integration configs | API configs, webhooks |
resource.ai |
AI model files, prompts | Model configs, prompt templates |
resource.modules |
External module dependencies | Third-party modules |
resource.runtime |
Runtime-generated files | Logs, temp files |
| File Type | Naming Pattern | Location |
|---|---|---|
| Core class | Obj<Name>.py |
factory.core/ |
| Service class | ObjService<Name>.py |
factory.service/ or factory.service/package.<client>/ |
| Report class | ObjReport<Name>.py |
factory.report/ |
| Export handler | ObjDataExport<Format>.py |
factory.export/ |
| Import handler | ObjDataImport<Format>.py |
factory.import/ |
| Documentation | Obj<Name>.md |
Same folder as .py file |
| SQL queries | Obj<Name>.yaml |
Same folder as .py file |
| Test file | test_Obj<Name>.py |
resource.test/pytests/factory.<module>/ |
| Shell script | <action>_<target>.sh |
resource.bin/ |
| HTML template | <name>.html + <name>.yaml |
resource.templates/ |
Is it Python code?
├── Yes → Is it a reusable component?
│ ├── Yes → Is it a base/utility class?
│ │ ├── Yes → factory.core/
│ │ └── No → What domain?
│ │ ├── Business logic → factory.service/
│ │ ├── Reports → factory.report/
│ │ ├── Web/forms → factory.web/
│ │ ├── Data export → factory.export/
│ │ ├── Data import → factory.import/
│ │ ├── AI/ML → factory.ai/
│ │ ├── Messaging → factory.conversation/
│ │ └── Other → appropriate factory.*
│ └── No → Is it client-specific?
│ ├── Yes → factory.<module>/package.<client>/
│ └── No → factory.<module>/
└── No → What type of file?
├── Configuration → resource.config/
├── Shell script → resource.bin/
├── Test data → resource.test/
├── Test code → resource.test/pytests/factory.<module>/
├── Documentation → resource.notes/
├── Web assets → resource.web/
├── HTML template → resource.templates/
└── Images → resource.images/