晨曦's Blog

select(func.count())
.select_from(table)
.where(
    table.c.status == 1,
)

func.COUNT(
    func.IF(
        user.c.province == "重庆",
        True,
        None,
    )
).label("province"),

select(["*"])
.select_from(table1)
.where(
    table1.c.status == 1,
    exists()
    .where(
        table1.c.id == table2.c.table1_id,
    ),
)

conditions = []
conditions.append(user.c.sex == 1)
select(["*"]).select_from(user).where(and_(*conditions))

user.update()
.where(
    user.c.id == 1
)
.values(score=user.c.score + 1)

#  升序
.order_by(user.c.id)
#  降序
.order_by(user.c.id.desc())

import datetime

'''
UTC转北京时间
'''

def utc_format(utc_time, utc_time_format='%Y-%m-%dT%H:%M:%S.000Z'):
    utc_datetime = datetime.datetime.strptime(utc_time, utc_time_format)
    local_datetime = utc_datetime + datetime.timedelta(hours=8)
    local_time_format = "%Y-%m-%d %H:%M:%S"
    local_time = local_datetime.strftime(local_time_format)
    return local_time

print(utc_format(utc_time="2021-10-26T09:34:31.000Z"))

async def aioredlock():
    # retry_count 重试次数，这里配置为1，杜绝进程重复去获取
    lock_manager = Aioredlock([{'host': 'localhost', 'port': 6379, 'db': 1}],retry_count=1)

    while True:

        # 每秒钟去检测锁的状态，如果锁被其他进程占用就跳过
        # lock_timeout 锁的时间为 65S，相当于进程拿到锁后最大65秒后释放
        # 业务逻辑执行 小于 65秒，业务逻辑执行完后当即释放锁
        # 每60秒执行一次 sleep 59s即可，因为检测已经sleep了1s

        await asyncio.sleep(1)
        if await lock_manager.is_locked("_ztsg_auto_generate_code"):
            #logger.info('The resource is already acquired')
            continue

        try:
            async with await lock_manager.lock("_ztsg_auto_generate_code", lock_timeout=65) as lock:

                assert lock.valid is True
                assert await lock_manager.is_locked("_ztsg_auto_generate_code") is True

                # 每60s钟运行一次任务
                await asyncio.sleep(59)

                logger.info('Start Tick! The time is '+str(os.getpid())+': %s' % datetime.now())
                # 业务代码
                logger.info('Stop Tick! The time is '+str(os.getpid())+': %s' % datetime.now())
        except LockAcquiringError:
            print('Something happened during normal operation. We just log it.')
        except LockError:
            print('Something is really wrong and we prefer to raise the exception')
        except Exception as e:
            print(e)
            print('Something else went wrong')

register_tortoise(
    app, db_url=db_url, modules={"models": ["app.models"]}, generate_schemas=False
)

register_tortoise(
    app,
    config={
        'connections': {
            'default': db_url
        },
        'apps': {
            'models': {
                "models": ["app.models"],
                'default_connection': 'default',
            }
        },
        "use_tz": False,
        "timezone": "Asia/Shanghai",
    },
    generate_schemas=False
)

import json
from base64 import b64encode, b64decode
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Random import get_random_bytes

class AesCbc:
    def __init__(self, key):
        self.bs = AES.block_size
        self.key = key
        self.mode = AES.MODE_CBC

    def encrypt(self, data):
        cipher = AES.new(self.key, self.mode)
        ct_bytes = cipher.encrypt(pad(data, self.bs))
        iv = b64encode(cipher.iv).decode('utf-8')
        ct = b64encode(ct_bytes).decode('utf-8')
        return json.dumps({'iv': iv, 'ciphertext': ct})

    def decrypt(self, json_input):
        try:
            b64 = json.loads(json_input)
            iv = b64decode(b64['iv'])
            ct = b64decode(b64['ciphertext'])
            cipher = AES.new(self.key, self.mode, iv)
            plaintext = unpad(cipher.decrypt(ct), self.bs)
            return plaintext.decode('utf-8')
        except (ValueError, KeyError) as err:
            print("Incorrect decryption ", err)
            return None

if __name__ == "__main__":
    data = "需要加密的数据".encode('utf-8')
    key = get_random_bytes(16)  # Bytes

    aes_cipher = AesCbc(key)

    encrypt_reuslt = aes_cipher.encrypt(data)
    print("密文: ", encrypt_reuslt)

    plaintext = aes_cipher.decrypt(encrypt_reuslt)
    print("原文: ", plaintext)

import json
from base64 import b64encode, b64decode
from Crypto.Cipher import AES
from Crypto.Random import get_random_bytes

class AesOcb:
    def __init__(self, key):
        self.bs = AES.block_size
        self.key = key
        self.mode = AES.MODE_OCB
        self.json_k = ['nonce', 'header', 'ciphertext', 'tag']

    def encrypt(self, header, data):
        header = header
        cipher = AES.new(self.key, self.mode)
        cipher.update(header)
        ciphertext, tag = cipher.encrypt_and_digest(data)
        json_v = [b64encode(x).decode('utf-8') for x in [cipher.nonce, header, ciphertext, tag]]
        return json.dumps(dict(zip(self.json_k, json_v)))

    def decrypt(self, json_input):
        try:
            b64 = json.loads(json_input)
            jv = {k: b64decode(b64[k]) for k in self.json_k}
            cipher = AES.new(self.key, self.mode, nonce=jv['nonce'])
            cipher.update(jv['header'])
            plaintext = cipher.decrypt_and_verify(jv['ciphertext'], jv['tag'])
            return plaintext.decode('utf-8')
        except (ValueError, KeyError) as err:
            # 解密错误
            print(err)
            return None

if __name__ == "__main__":
    data = "需要加密的数据".encode('utf-8')
    key = get_random_bytes(16)  # Bytes
    header = b'header'

    aes_cipher = AesOcb(key)

    encrypt_reuslt = aes_cipher.encrypt(header, data)
    print("密文: ", encrypt_reuslt)

    plaintext = aes_cipher.decrypt(encrypt_reuslt)
    print("原文: ", plaintext)