[Python] Mysterious Memory hike in Flask Admin

we observed a significant memory spike (~235MB) when editing and saving an article model using flask admin, even though the on_model_change() and after_model_change() methods perform minimal logic.

what i did to identify this:

i added memory profiling using psutil around all admin methods (edit_form, on_model_change, after_model_change and get_list) and i observed:

edit_form: stable memory ~135MB

on_model_change: memory jumped to ~375MB (the memory spike occurs before on_model_change() runs)

on_model_change() body was effectively empty for that test model

after_model_change: added almost no memory (~0.1MB)

model.index() (Elasticsearch update): used less than 1MB



experiements i did:

i tried calling garbage collector as an app hook like below and gc was able make some memory free

@app.after_request
def invoke_gc(response):
gc.collect()
db.session.expunge_all()
return response


we applied the same profiling to another lightweight model (CategoryAdmin) to confirm if this is a flask admin level issue and not isolated to the article model and we observed a smaller but noticeable memory increase (e.g., ~10–20MB), despite no heavy relationships or logic. given that the article model includes multiple relationships and the ArticleAdmin class uses many form widgets. these may be contributing to increased memory usage during form handling. i would really appreciate feedback from you before proceeding further.

article model class

• Please note that the article table is heavily bound with other table relationships

class Article(db.Model):
__tablename__ = "articles"

def __repr__(self):
return "Article(%s)" % self.id

def __str__(self):
return self.title

id = db.Column(db.Integer(), primary_key=True, nullable=False)
created = db.Column(db.DateTime(), default=datetime.utcnow, nullable=False)
updated = db.Column(db.DateTime(), onupdate=datetime.utcnow)
indexed = db.Column(db.DateTime())
enabled = db.Column(db.Boolean(), nullable=False, default=True)
published = db.Column(
db.DateTime(), nullable=False, default=datetime.now
) # Not UTC
public = db.Column(db.Boolean(), nullable=False, default=False)
title = db.Column(db.String(300), nullable=False, unique=True)
url = db.Column(db.String(300), nullable=False, unique=True)
snippet_title = db.Column(db.String(300))
author = db.Column(db.String(100))
short_description = db.Column(db.Text())
content = db.Column(db.Text())
content_markdown = db.Column(db.Text())
cpd_points = db.Column(db.String(20))
show_comments = db.Column(db.Boolean(), nullable=False, default=True)
enable_comments = db.Column(db.Boolean(), nullable=False, default=True)
views = db.Column(db.Integer(), nullable=False, default=0)
acl = db.Column(db.String(250))
linkedin_link = db.Column(db.String(1000))
twitter_link = db.Column(db.String(1000))
email_link = db.Column(db.String(1000))
legacy = db.Column(db.Boolean(), nullable=False, default=False)
fullwidth = db.Column(db.Boolean(), nullable=False, default=False)
session_limit = db.Column(db.Integer())
session_limit_message = db.Column(db.Text())
redirect_url = db.Column(db.String(300))
show_similar_articles = db.Column(db.Boolean(), nullable=False, default=True)
suggest = db.Column(db.Boolean(), nullable=False, default=True)
budsids_to_subscribe = db.Column(db.String(200))
show_top_bar = db.Column(db.Boolean(), nullable=False, default=True)
sharing_platform_title = db.Column(db.String(1000))
sharing_platform_body = db.Column(db.Text())

thumbnail_id = db.Column(
db.Integer(),
db.ForeignKey("images.id", onupdate="CASCADE", ondelete="RESTRICT"),
)

@property
def comment_count(self):
return len(self.comments)

@property
def format_names(self):
return (
c.name
for c in self.categories
if c.name in ("Read", "Watch", "Listen", "Attend")
)

@property
def domain_names(self):
return (c.name for c in self.categories if c.domain)


def index(self, delete=False):
import tracemalloc
import time
from elasticsearch import Elasticsearch
try:
tracemalloc.start()
start_time = time.time()

es = Elasticsearch(app.config["ELASTICSEARCH_HOST"])
index_name = app.config.get("ES_ARTICLE_INDEX", "article")

if delete:
print(f"[Index] Deleting article ID {self.id}")
es.delete(index=index_name, id=self.id, ignore=[404])
return

print(f"[Index] Partial update for article ID {self.id}")

# Prepare categories and entities
categories = [c.name for c in self.categories]
entities = app.config.get("ENTITIES", [])

# Prepare dynamic values
program_names = []
competency = []
entity_list = []

for category in self.categories:
if category.program:
program_names.append(category.name)
if category.name.startswith("Competency - "):
competency.append(category.name.replace("Competency - ", ""))
for entity in entities + ["FASEA", "FPA"]:
if category.name.endswith(f"({entity})"):
modified = {"FASEA": "Legislated (AU)", "FPA": "FAAA"}.get(entity, entity)
if modified not in entity_list:
entity_list.append(modified)

payload = {
"title": self.title,
"snippet_title": self.snippet_title,
"author": self.author,
"short_description": self.short_description,
"content": self.content[:500000] if self.content else "",
"categories": ",".join(categories),
"published": self.published,
"year": str(self.published.strftime("%Y")),
"url": f"/article/{self.id}/{self.url}",
"cpd_points": self.cpd_points,
"thumbnail": self.thumbnail.filename if self.thumbnail else None,
"comment_count": self.comment_count,
"format_names": list(self.format_names),
"domain_names": list(self.domain_names),
"program_names": program_names,
"competency": competency,
"entities": entity_list,
"suggest": self.suggest,
"enabled": self.enabled
}

# Remove None values from payload to avoid ES rejection
filtered_payload = {k: v for k, v in payload.items() if v is not None}

# Perform partial update
es.update(
index=index_name,
id=self.id,
body={
"doc": filtered_payload,
"doc_as_upsert": True
}
)

now = datetime.utcnow()
self.indexed = now
self.updated = now
db.session.commit()

except Exception as e:
print(f"[Index] ❌ Failed to partially update article {self.id}: {str(e)}")
raise

finally:
current, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
elapsed = time.time() - start_time
print(
f"[Index] Finished article {self.id} | Time: {elapsed:.2f}s | Mem: {current / 10 ** 6:.2f}MB | Peak: {peak / 10 ** 6:.2f}MB")

def add_view(self):
view_id = None
if not app.config.get("READONLY", None) and current_user.is_authenticated:
self.views += 1
for i in self.images:
i.image.impressions += 1
if current_user.is_authenticated:
article_view = ArticleView(
user_id=current_user.person_id, article_id=self.id, active=True
)
db.session.add(article_view)
db.session.flush()
view_id = article_view.id
db.session.commit()
return view_id

@property
def content_processed(self):
content = self.content
if current_user.is_authenticated:
content = content.replace("%7B%7BUSER_HASH%7D%7D", "{{USER_HASH}}")
content = content.replace(
"{{USER_HASH}}", current_user.get_attribute("hash")
)
content = content.replace("%7B%7BUSER_UUID%7D%7D", "{{USER_UUID}}")
content = content.replace(
"{{USER_UUID}}", current_user.get_attribute("uuid")
)
return content

@property
def content_truncated(self):
return truncate(self.content_processed, 700, "...")

@cached_property
def related_articles(self):
combined = list(self.manually_related_articles)
if self.author:
other_by = Article.query.filter(Article.author == self.author).filter(
Article.id != self.id
)
if combined:
other_by = other_by.filter(
db.not_(Article.id.in_([a.id for a in combined]))
)
other_by = other_by.order_by(Article.published.desc()).limit(10).all()
combined += list(other_by)
combined = [a for a in combined if a.enabled]
return sorted(combined, key=attrgetter("published"), reverse=True)[0:15]

@property
def link(self):
link = "/article/%s/%s" % (self.id, self.url)
return link

@property
def active_session_count(self):
juncture = datetime.utcnow() - timedelta(seconds=15)
return self.article_views.filter(ArticleView.last_seen >= juncture).count()

@property
def is_saved(self):
if current_user.is_authenticated:
return (
self.saved_items.filter(
SavedItem.person_id == current_user.person_id
).count()
> 0
)


article admin class

class ArticleAdmin(BaseModelView):
create_template = "admin/bws-article-create.html"
edit_template = "admin/bws-article-edit.html"

page_size = 30

column_list = (
"published",
"enabled",
"title",
"author",
"views",
)
column_searchable_list = ("title", "author", "snippet_title")
column_labels = dict(title="Article Title")
column_filters = ("author", "title")
column_default_sort = ("published", True)
column_formatters = dict(
title=lambda v, c, m, p: Markup('<a href="%s">%s</a>') % (m.link, m.title)
)

form_overrides = {"acl": SelectMultipleField}
form_excluded_columns = (
"created",
"updated",
"indexed",
"comments",
"views",
"comment_notifications",
"article_views",
"linkedin_link",
"twitter_link",
"email_link",
)
form_args = dict(
title=dict(label="Title"),
content=dict(label="Old Content"),
acl=dict(label="Access Group", choices=[]),
enable_comments=dict(label="Enable Commenting"),
cpd_points=dict(label="CE Hours"),
public=dict(label="Public Article"),
manually_related_articles=dict(label="Related Articles"),
redirect_url=dict(validators=[URL()]),
show_similar_articles=dict(label="Similar Articles"),
budsids_to_subscribe=dict(label="BUDSIDs to subscribe"),
)
form_widget_args = {
"short_description": {"style": "height: 60px;"},
"content": {"style": "height: 300px;"},
"enabled": {"data-toggle": "toggle"},
"public": {"data-toggle": "toggle"},
"show_comments": {"data-toggle": "toggle"},
"enable_comments": {"data-toggle": "toggle"},
"legacy": {"data-toggle": "toggle"},
"fullwidth": {"data-toggle": "toggle"},
"show_similar_articles": {"data-toggle": "toggle"},
"suggest": {"data-toggle": "toggle"},
"show_top_bar": {"data-toggle": "toggle"},
}
form_rules = [
"enabled",
"published",
"public",
"title",
"url",
"snippet_title",
"author",
"short_description",
"legacy",
"fullwidth",
rules.Field("content", "content"),
rules.Field("content_markdown", "content_markdown"),
"cpd_points",
"show_comments",
"enable_comments",
"acl",
# "email_link",
# "linkedin_link",
# "twitter_link",
"thumbnail",
"show_similar_articles",
"suggest",
"manually_related_articles",
"categories",
"images",
"session_limit",
"session_limit_message",
"redirect_url",
"budsids_to_subscribe",
"show_top_bar",
"sharing_platform_title",
"sharing_platform_body",
]

inline_model_form_converter = CustomInlineModelConverter
inline_models = [
(
ArticleImage,
dict(
form_excluded_columns=("created", "updated"),
form_widget_args={
"image": {"column_class": "col-md-6"},
"sort": {"column_class": "col-md-2"},
},
),
)
]

def _profile(func):
def wrapper(*args, **kwargs):
process = psutil.Process(os.getpid())
start_mem = process.memory_info().rss # in bytes
start_time = time.time()

try:
return func(*args, **kwargs)
finally:
end_time = time.time()
end_mem = process.memory_info().rss
diff = (end_mem - start_mem) / (1024 * 1024)
print(
f"[PROFILE] {func.__name__} → Time: {end_time - start_time:.4f}s | ΔMemory: {diff:.2f}MB | Now: {end_mem / (1024 * 1024):.2f}MB")

return wrapper

@_profile
def get_one(self, id):
return (
self.session.query(self.model)
.options(lazyload("*")) # Ensure edit view also avoids loading relationships
.get(id)
)

@_profile
def get_list(self, page, sort_field, sort_desc, search, filters, execute=True, page_size=None):
return super(ArticleAdmin, self).get_list(page, sort_field, sort_desc, search, filters, execute, page_size)


@_profile
def create_form(self, obj=None):
form = super(ArticleAdmin, self).create_form(obj=obj)
form.acl.choices = app.config.get("ACLS", [])
form.acl.widget = Select2Widget(multiple=True)
return form

@_profile
def edit_form(self, obj=None):
process, start_mem, end_mem = None, 0, 0
process = psutil.Process(os.getpid())
start_mem = process.memory_info().rss

form = super(ArticleAdmin, self).create_form(obj=obj)

form.acl.choices = app.config.get("ACLS", [])
form.acl.widget = Select2Widget(multiple=True)

end_mem = process.memory_info().rss
print(
f" populate obj : {start_mem / (1024 * 1024):.2f}MB | Now: {end_mem / (1024 * 1024):.2f}MB")

return form

@_profile
def on_form_prefill(self, form, id):
model = self.session.query(self.model).get(id)
if model.budsids_to_subscribe:
session["budsids_to_unsubscribe"] = model.budsids_to_subscribe.split(",")
if model.acl:
acl_list = model.acl.strip("{}")
if acl_list:
form.acl.data = acl_list.split(",")

@_profile
def on_model_change(self, form, model, is_created):
process, start_mem, end_mem = None, 0, 0
process = psutil.Process(os.getpid())
start_mem = process.memory_info().rss
if model.budsids_to_subscribe:
invalid_ids = []
ids_list = model.budsids_to_subscribe.split(",")
for id in ids_list:
r = requests.get(
"%s/json/search_people/" % app.config["BUDS_URL"],
params={"key": app.config["BUDS_KEY"], "ids": id.strip()},
)
if r.status_code == 200:
data = r.json()
if not data:
invalid_ids.append(id.strip())
else:
invalid_ids.append(id.strip())

if invalid_ids:
message = "Could not find a person for the BudsID(s): {}".format(
", ".join(invalid_ids)
)
raise ValidationError(message)
else:
budsids_to_unsubscribe = session.pop("budsids_to_unsubscribe", None)
if budsids_to_unsubscribe:
for id in budsids_to_unsubscribe:
existing = (
CommentNotification.query.filter(
CommentNotification.article == model
)
.filter(CommentNotification.person_id == id)
.first()
)
if existing:
db.session.delete(existing)
db.session.commit()

for id in ids_list:
cn = CommentNotification()
cn.article = model
cn.person_id = id
db.session.add(cn)
db.session.commit()
else:
budsids_to_unsubscribe = session.pop("budsids_to_unsubscribe", None)
if budsids_to_unsubscribe:
for id in budsids_to_unsubscribe:
existing = (
CommentNotification.query.filter(
CommentNotification.article == model
)
.filter(CommentNotification.person_id == id)
.first()
)
if existing:
db.session.delete(existing)
db.session.commit()
end_mem = process.memory_info().rss
print(
f" on model change : {start_mem / (1024 * 1024):.2f}MB | Now: {end_mem / (1024 * 1024):.2f}MB")

Continue reading...