kw_tools/tests/redistest.py

import time
import json
import random
import dataset
import redis
import string
import logging
import sys
from logging.handlers import SysLogHandler
import socket
#_LOG_SERVER = ('172.105.219.42', 514)
_LOG_SERVER = ('hhh.ptt.cx', 514)

logger = logging.getLogger('Jared')
logger.setLevel(logging.FATAL)
handler1 = SysLogHandler(
        address=_LOG_SERVER,socktype=socket.SOCK_DGRAM)
formatter1 = logging.Formatter('%(module)s: %(process)s - %(name)s - %(levelname)s - %(lineno)s - %(message)s')
handler1.setFormatter(formatter1)
logger.addHandler(handler1)
#logger.fatal()
###logging.setLoggerClass('cc')
r = logging.makeLogRecord({'msg': '測試訊息','filename':'fname','funcName':'fcname','lineno':1234,'threadName':'thname','process':54321})
handler1.handle(r)
#logger.

#logger.fatal('ttt')
#logger.info('this is a test')
d = {'clientip': '192.168.0.999', 'user': 'fbloggs'}
logger.fatal('測試一下',extra=d,exc_info=1)
#logger.debug('中文測試')

sys.exit()

def id_generator(size=6, chars=string.ascii_uppercase + string.digits):
    return ''.join(random.choice(chars) for _ in range(size))

r = redis.Redis(host='db.ptt.cx', port=6379, db=1,password='choozmo9')


#r.set('hbeat:20220319', 1, ex=90)
for key in r.scan_iter("hbeat:*"):
    print(key)
    # delete the key

#r.get()

#    r.delete(key)

#r = redis.Redis(host='db.ptt.cx', port=6379, db=1)
#len=r.llen('test')
#lst=r.lrange('test',0,len-1)
#print(lst)
#r.delete('test')

#for i in range(5):
#    js={'name':id_generator(),'id':random.randint(111111,999999)}
#    r.lpush('test',json.dumps(js))

#p = r.pubsub(ignore_subscribe_messages=True)