ObjNewsFeed is a class for managing social media-style news feeds with support for hashtags, mentions, HTML processing, and database storage. It provides CRUD operations for feed items with features like batch insertions, text-to-HTML conversion, and profile-based filtering.
Module: factory.core/ObjNewsFeed.py
Inherits from: Base class with database connectivity
Test file: resource.test/pytests/factory.core/test_ObjNewsFeed.py
Database table: data_newsfeed
This module enables developers to:
data_newsfeed tableescape_sql()| Field | Type | Description |
|---|---|---|
| guid | UUID | Unique identifier (generated) |
| shortcode | String | Short unique code for URL/reference |
| image_path | String | Path to associated image |
| feedtexl | Text | Original plain text content |
| feedhtml | HTML | Processed HTML version |
| profile_name | String | Associated profile/user name |
| post_timestamp | Datetime | When item was posted |
| download_timestamp | Datetime | When item was downloaded/imported |
__init__(DB=None)Initializes the ObjNewsFeed instance.
Parameters:
DB (optional): Database connection object_process_feed_text_to_html(text: str) -> strConverts plain text to HTML with hashtag and mention formatting.
Parameters:
text (str): Plain text to processReturns:
Processing Rules:
#hashtag to <b>#hashtag</b>@mention to <b>@mention</b><p> tagsExamples:
feed = ObjNewsFeed()
# Hashtags
feed._process_feed_text_to_html("This is #awesome")
# Returns: '<p>This is <b>#awesome</b></p>'
# Mentions
feed._process_feed_text_to_html("Hello @user")
# Returns: '<p>Hello <b>@user</b></p>'
# Multiple paragraphs
feed._process_feed_text_to_html("Paragraph 1\n\nParagraph 2")
# Returns: '<p>Paragraph 1</p><p>Paragraph 2</p>'
# XSS prevention
feed._process_feed_text_to_html("<script>alert('xss')</script>")
# Returns: '<p><script>alert('xss')</script></p>'
insert_new_feed_item(guid, shortcode, image_path, feedtexl, feedhtml, profile_name, post_timestamp, download_timestamp)Inserts a new feed item into the database.
Parameters:
guid (str): Unique identifiershortcode (str): Short unique codeimage_path (str): Path to imagefeedtexl (str): Plain text contentfeedhtml (str): HTML contentprofile_name (str): Profile namepost_timestamp (str): Post datetimedownload_timestamp (str): Download datetimeExample:
feed = ObjNewsFeed()
feed.insert_new_feed_item(
guid="uuid-123",
shortcode="post-abc",
image_path="/images/post.jpg",
feedtexl="Hello world #test",
feedhtml="<p>Hello world <b>#test</b></p>",
profile_name="johndoe",
post_timestamp="2026-02-04 10:00:00",
download_timestamp="2026-02-04 10:05:00"
)
item_exists(shortcode: str) -> boolChecks if a feed item with the given shortcode exists.
Parameters:
shortcode (str): Shortcode to checkReturns:
True if item exists, False otherwiseupsert_feed_item(guid, shortcode, image_path, feedtexl, feedhtml, profile_name, post_timestamp, download_timestamp) -> boolInserts a new item or updates an existing one.
Parameters:
insert_new_feed_item()Returns:
True if item was inserted (new)False if item was updated (existing)Example:
feed = ObjNewsFeed()
# First call - inserts
result = feed.upsert_feed_item(...) # Returns True
# Second call with same shortcode - updates
result = feed.upsert_feed_item(...) # Returns False
insert_feed_items_batch(items: List[Dict]) -> intBatch inserts multiple feed items for performance.
Parameters:
items (List[Dict]): List of feed item dictionariesReturns:
Example:
feed = ObjNewsFeed()
items = [
{
"guid": "guid1",
"shortcode": "post1",
"image_path": "/images/1.jpg",
"feedtexl": "Post 1",
"feedhtml": "<p>Post 1</p>",
"profile_name": "user1",
"post_timestamp": "2026-02-04 10:00:00",
"download_timestamp": "2026-02-04 10:00:00"
},
{
"guid": "guid2",
"shortcode": "post2",
# ... more fields
}
]
count = feed.insert_feed_items_batch(items)
# Returns: 2
create_feed_item(text, profile_name, shortcode=None, image_path="", post_timestamp=None, download_timestamp=None, process_html=True) -> strHigh-level method to create a feed item with automatic processing.
Parameters:
text (str): Plain text content (required)profile_name (str): Profile name (required)shortcode (str, optional): Custom shortcode (auto-generated if None)image_path (str, optional): Image path (default: "")post_timestamp (str, optional): Post time (auto-generated if None)download_timestamp (str, optional): Download time (defaults to post_timestamp)process_html (bool, optional): Whether to process text to HTML (default: True)Returns:
Example:
feed = ObjNewsFeed()
# Minimal usage
guid = feed.create_feed_item(
text="Hello world #awesome @johndoe",
profile_name="myprofile"
)
# With custom shortcode
guid = feed.create_feed_item(
text="My post",
profile_name="myprofile",
shortcode="custom-code-123",
image_path="/images/photo.jpg"
)
# Without HTML processing
guid = feed.create_feed_item(
text="Raw content",
profile_name="myprofile",
process_html=False
)
get_feed_item(shortcode: str) -> DictRetrieves a single feed item by shortcode.
Parameters:
shortcode (str): Shortcode to retrieveReturns:
None if not foundExample:
feed = ObjNewsFeed()
item = feed.get_feed_item("post-123")
if item:
print(item['feedtexl'])
print(item['profile_name'])
get_feed_items(limit=20, offset=0, profile_name=None) -> List[Dict]Retrieves multiple feed items with optional filtering.
Parameters:
limit (int, optional): Maximum items to return (default: 20)offset (int, optional): Number of items to skip (default: 0)profile_name (str, optional): Filter by profile nameReturns:
Example:
feed = ObjNewsFeed()
# Get latest 10 items
items = feed.get_feed_items(limit=10)
# Pagination - second page
items = feed.get_feed_items(limit=10, offset=10)
# Filter by profile
items = feed.get_feed_items(profile_name="johndoe")
# Profile filter with pagination
items = feed.get_feed_items(limit=20, offset=40, profile_name="johndoe")
delete_feed_item(shortcode: str) -> boolDeletes a feed item by shortcode.
Parameters:
shortcode (str): Shortcode of item to deleteReturns:
True if item was deletedFalse if item doesn't existExample:
feed = ObjNewsFeed()
if feed.delete_feed_item("old-post-123"):
print("Item deleted successfully")
else:
print("Item not found")
update_feed_item(shortcode: str, text=None, image_path=None, profile_name=None) -> boolUpdates specific fields of an existing feed item.
Parameters:
shortcode (str): Shortcode of item to update (required)text (str, optional): New text (also updates feedhtml)image_path (str, optional): New image pathprofile_name (str, optional): New profile nameReturns:
True if update succeededFalse if item doesn't existNotes:
text is updated, feedhtml is automatically regeneratedTrue even if no fields are specified (no-op)Example:
feed = ObjNewsFeed()
# Update text only
feed.update_feed_item("post-123", text="Updated content #new")
# Update multiple fields
feed.update_feed_item(
shortcode="post-123",
text="New text",
image_path="/images/new.jpg"
)
# Update profile only
feed.update_feed_item("post-123", profile_name="newuser")
from ObjNewsFeed import ObjNewsFeed
feed = ObjNewsFeed()
# Create a simple feed item
guid = feed.create_feed_item(
text="Just deployed our new feature! #ProductLaunch @engineering",
profile_name="company"
)
print(f"Created feed item: {guid}")
from ObjNewsFeed import ObjNewsFeed
import datetime
feed = ObjNewsFeed()
# Import social media posts
posts = fetch_social_media_posts() # Your API call
items = []
for post in posts:
items.append({
"guid": post.id,
"shortcode": post.shortcode,
"image_path": post.image_url,
"feedtexl": post.text,
"feedhtml": feed._process_feed_text_to_html(post.text),
"profile_name": post.author,
"post_timestamp": post.created_at,
"download_timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
# Batch insert for performance
count = feed.insert_feed_items_batch(items)
print(f"Imported {count} posts")
from ObjNewsFeed import ObjNewsFeed
from fastapi import FastAPI, Query
app = FastAPI()
feed = ObjNewsFeed()
@app.get("/feed")
def get_feed(
limit: int = Query(20, le=100),
offset: int = Query(0, ge=0),
profile: str = None
):
items = feed.get_feed_items(
limit=limit,
offset=offset,
profile_name=profile
)
return {"items": items, "count": len(items)}
@app.post("/feed")
def create_post(text: str, profile: str, image: str = ""):
guid = feed.create_feed_item(
text=text,
profile_name=profile,
image_path=image
)
return {"guid": guid, "status": "created"}
@app.delete("/feed/{shortcode}")
def delete_post(shortcode: str):
success = feed.delete_feed_item(shortcode)
return {"status": "deleted" if success else "not_found"}
The module has comprehensive test coverage for all functionality:
| Test Case | Description | Status |
|---|---|---|
test_process_feed_text_to_html_hashtags |
Hashtag formatting | ✓ |
test_process_feed_text_to_html_mentions |
Mention formatting | ✓ |
test_process_feed_text_to_html_mixed |
Mixed content with paragraphs | ✓ |
test_process_feed_text_to_html_empty_input |
Empty string handling | ✓ |
test_process_feed_text_to_html_escapes_html |
XSS prevention | ✓ |
test_insert_new_feed_item |
Basic insertion | ✓ |
test_item_exists_returns_true |
Existence check (true) | ✓ |
test_item_exists_returns_false |
Existence check (false) | ✓ |
test_upsert_feed_item_inserts_when_not_exists |
Upsert insert path | ✓ |
test_upsert_feed_item_updates_when_exists |
Upsert update path | ✓ |
test_insert_feed_items_batch_empty_list |
Empty batch handling | ✓ |
test_insert_feed_items_batch_multiple_items |
Batch insertion | ✓ |
test_create_feed_item |
High-level creation | ✓ |
test_create_feed_item_with_custom_shortcode |
Custom shortcode | ✓ |
test_create_feed_item_without_html_processing |
Raw text mode | ✓ |
test_get_feed_item |
Single item retrieval | ✓ |
test_get_feed_item_not_found |
Not found handling | ✓ |
test_get_feed_items |
Multiple item retrieval | ✓ |
test_get_feed_items_with_profile_filter |
Profile filtering | ✓ |
test_delete_feed_item_success |
Successful deletion | ✓ |
test_delete_feed_item_not_found |
Delete non-existent | ✓ |
test_update_feed_item_text |
Text update | ✓ |
test_update_feed_item_not_found |
Update non-existent | ✓ |
test_update_feed_item_no_changes |
No-op update | ✓ |
Test file location: resource.test/pytests/factory.core/test_ObjNewsFeed.py
Run tests:
pytest resource.test/pytests/factory.core/test_ObjNewsFeed.py -v
The module automatically escapes HTML in user input:
# Malicious input
text = "<script>alert('xss')</script>"
# Safe output
html = feed._process_feed_text_to_html(text)
# Result: '<p><script>alert('xss')</script></p>'
All SQL parameters are escaped via escape_sql():
# User input
shortcode = "post'; DROP TABLE data_newsfeed; --"
# Safe query
feed.get_feed_item(shortcode) # Escaped before SQL execution
insert_feed_items_batch() instead of individual insertsThe module uses YAML-based query templates:
insert_feed_item - Insert single iteminsert_feed_items_batch - Batch insertupdate_feed_item_full - Full update (upsert)update_feed_item - Partial updateget_feed_item - Get by shortcodeget_feed_items - Get multiple itemsget_feed_items_by_profile - Get by profiledelete_feed_item - Delete by shortcodeitem_exists - Check existenceObjWorkflow.py - Can trigger feed creation via workflowsObjApi.py - API endpoints for feed managementServeWebsite.py - Web interface for displaying feedsObjects.Object().get_uuid() for GUID generationprocess_html=True in create_feed_item()