Repository URL to install this package:
|
Version:
4.0.1 ▾
|
import typing as t
import sqlalchemy as sa
from sarus_statistics.ops.utils import rescale_weights_sql
# pylint: disable=too-many-arguments, too-many-locals
def links(
session: sa.orm.Session,
table: sa.sql.FromClause,
col_name: str,
user_col: str,
weight_col: str,
private_col: str,
mult_fk: float,
size_fk: int,
max_categories: t.Optional[int] = None,
) -> sa.orm.Query:
if not max_categories:
max_categories = size_fk
private_data = session.query(table)
weighted_data = rescale_weights_sql(
session,
private_data.subquery(),
col_name,
user_col=user_col,
private_col=private_col,
weight_col=weight_col,
max_multiplicity=mult_fk,
)
weighted_col = getattr(weighted_data.c, col_name)
private_hist = (
session.query(
weighted_col.label('data'),
sa.func.count().label('count_'),
)
.group_by(weighted_col)
.subquery()
)
prevalences = session.query(
private_hist.c.count_.label('data'),
sa.func.count().label('count_'),
).group_by(private_hist.c.count_)
return t.cast(sa.orm.Query, prevalences)