One post tagged with "数据库"

class User(BaseModel):
    __tablename__ = 'user'
    username = db.Column(db.string(64), nullable=False, index=True)
    org_id = db.Column(db.SMALLINT, nullable=False, comment='组织id')
    delete_id = db.Column(db.INTEGER, nullable=False, default=0)

    __table_args__ = (
        db.UniqueConstraint('username', 'delete_id', name='_username_delete_id_uc'),
    )

user_roles = db.Table('sys_user_role',
    db.Column('id', db.Integer, primary_key=True),
    db.Column('user_id', db.Integer, nullable=False),
    db.Column('role_id', db.Integer, nullable=False)
)


@bp.post('/user_role')
@parser.use_args({
    'role_id': fields.Int(required=True),
    'user_id': fields.Int(required=True)
}, location='json')
@role_required()
def set_user_role(user, args):
    stmt = user_roles.insert().values(**args).returning(user_roles)
    ret = db.session.execute(stmt)
    ret = ret.mappings().one_or_none()
    db.session.commit()

    return response_success(data=dict(ret))

# 更新历史分配记录

UPDATE_POST_SCORE = """
    UPDATE post_community SET score = tmp.score
    FROM (VALUES (%s, %s)) AS tmp(id, score)
    WHERE post_community.post_id=tmp.id;
"""

from psycopg2.extras import execute_batch
execute_batch(cursor, UPDATE_POST_SCORE, result)

cursor = self.conn.connection.cursor(name="named_cursor")
cursor.itersize = 5000
cursor.execute(SELECT_POST_COUNTS)
data = cursor.fetchmany(20000)

now = DT.now_time()
while data:
    result = map(lambda x: (
        x[0],
        (x[1] + 2 * x[2] + 3 * x[3]) / (6 * ((round((now - x[-1].replace(tzinfo=now.tzinfo)).total_seconds() / 3600 + 2)) ** 1.8))
    ), data)
    execute_batch(self.cursor, UPDATE_POST_SCORE, result)
    data = cursor.fetchmany(20000)

from dateutil.relativedelta import relativedelta


def part_table_by_month(sub_table):

    """按月分区表"""
    now = DT.cur_dt()
    next = now + relativedelta(months=1)  # 下一个月
    next_next = next + relativedelta(months=1)  # 下下个月

    part_name = f"{sub_table}_{next.year}{next.month:02d}"
    start = DT.str2ts(f"{next.year}-{next.month}-01", "%Y-%m-%d")
    end = DT.str2ts(f"{next_next.year}-{next_next.month}-01", "%Y-%m-%d") - 1
    try:
        part_ret = part_table(sub_table, start, end)
        logger.info(f"创建分区表成功: {part_ret}")
    except Exception as e:
        logger.critical(f"创建分区表失败: {part_name}, 失败原因：{str(e)}")


def part_table(sub_table: str, start, end) -> str:
    """
    按月创建分区表（当月触发，创建下个月的）
    :param end: 分区结束范围
    :param start: 分区起始范围
    :param sub_table: 主表名
    :return: 分区表名
    """

    sql_command = f"""CREATE TABLE IF NOT EXISTS {sub_table} PARTITION OF {sub_table} FOR VALUES FROM (:start) TO (:end);"""
    try:
        db.session.execute(text(sql_command), {"start": start, "end": end})
        db.session.commit()
        return sub_table
    except Exception as e:
        db.session.rollback()
        raise e