devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

MySql --sqlalchemy---mongo示例用法_sqlalchemy连接mango

作者：菜鸟追梦旅行 | 2024-05-23 14:27:59

赞

踩

sqlalchemy连接mango
  1. from sqlalchemy.ext.declarative import declarative_base
  2. from sqlalchemy import Column, Integer, String, UniqueConstraint, Index, BigInteger, Date, Text, not_
  3. from sqlalchemy.orm import sessionmaker
  4. from sqlalchemy import create_engine
  5. import mongo_connect
  6. from mongo_connect import MongoDao
  7. from mongo_connect import url,db,coll_name
  8. import time
  9.  
  10. # from utils.dataetl_config import notice_host, notice_port, notice_user, notice_pwd, notice_db
  11.  
  12. host="192.168.41.128"
  13. port=3306
  14. user="root"
  15. passwd="mysql"
  16. db="collect_resultdb"
  17.  
  18. engine = create_engine("mysql+pymysql://%s:%s@%s:%s/%s?charset=utf8&autocommit=true" % (user,passwd,host,port,db),
  19.                        max_overflow=20,pool_size=10)
  20.  
  21. # from wechat.models import SystemConst, ResearchReport
  22. Session = sessionmaker(bind=engine)
  23. session = Session()
  24.  
  25. Base = declarative_base()
  26.  
  27. class Notice(Base):
  28.     """
  29.     该表收录了国内主流券商微信公众号研报内容
  30.     """
  31.     __tablename__ = 'collect_gtja'
  32.     # ID = models.BigIntegerField(primary_key=True)
  33.     info_id = Column(String(255), primary_key=True)
  34.     datetime = Column(String(64))
  35.     title = Column(String(255))
  36.     file_name = Column(String(255))
  37.     URL =  Column(String(255))
  38.     source = Column(String(255))
  39.  
  40.     content =  Column(Text) # models.CharField(max_length=50, primary_key=True)
  41.     file_path = Column(Text)
  42.     file_attach = Column(Text)
  43.     summary = Column(Text)
  44.  
  45.     stock_name = Column(String(255))
  46.     stock_code = Column(String(255))
  47.     news_type = Column(String(255))
  48.     info_type = Column(String(255))
  49.     author = Column(String(255))
  50.  
  51.     image_attach = Column(Text)
  52.     status = Column(String(10))
  53.  
  54.     file_id = Column(String(64))
  55.  
  56.     __table_args__ = (
  57.     UniqueConstraint('info_id'),
  58.        Index('info_id'),
  59.     )
  60.  
  61.     def query_by_update_time(self,offset,now,status="1"):
  62.         session = Session()
  63.         # dt_s = datetime.now().date()
  64.         reports = session.query(Notice).filter(
  65.             # Notice.datetime.between(offset,now),
  66.             Notice.status == status
  67.         )
  68.         return session,reports
  69.  
  70.     def create(self):
  71.         Session = sessionmaker(bind=engine)
  72.         session = Session()
  73.         session.add(self)
  74.         session.commit()
  75.  
  76.     def query_by_id(self,id):
  77.         session = Session()
  78.         reports = session.query(Notice).filter(
  79.             # Notice.datetime.between(offset,now),
  80.             Notice.info_id == id
  81.         )
  82.         return session,reports
  83.  
  84.     def query_by_status(self):
  85.         session = Session()
  86.         # dt_s = datetime.now().date()
  87.         reports = session.query(Notice).filter(
  88.             # Notice.datetime.between(offset,now),
  89.             Notice.status == 1
  90.         )
  91.         return session,reports
  92.  
  93. if __name__ == '__main__':
  94.  
  95.     notice = Notice()
  96.     mongodao = MongoDao(mongo_connect.url,mongo_connect.db,mongo_connect.coll_name)
  97.     mongodao.collection
  98.     se,results = notice.query_by_status()
  99.     # print(results,"$"*20)
  100.     # print(type(results))
  101.     # print(results.values())
  102.     # print(results.all())
  103.     # for l in results.all():
  104.     # # l = results.first()
  105.     #     print(l.author)
  106.     for result in results.all():
  107.         spider_data = result
  108.         data = {}
  109.         data["update_time"] = time.time()
  110.         data["title"] = spider_data.title
  111.         data["content"] = ""
  112.         data["html"] = ""
  113.  
  114.         data["url"] = spider_data.URL
  115.  
  116.         data["source_id"] = ""
  117.         data["summary"] = ""
  118.  
  119.         data["content_time"] = spider_data.datetime
  120.  
  121.         # data["info_type"] = InfoType.notice.value
  122.  
  123.         data["source"] = spider_data.source
  124.         data["source_channel"] = ""
  125.  
  126.         # data["channel"] = ChannelType.spider.value
  127.         data["stock_name"] = spider_data.stock_name
  128.         data["stock_code"] = spider_data.stock_code
  129.  
  130.         # data["file_type"] = DataType.pdf.value
  131.         # data["keywords"] = spider_data["summary"]
  132.         data["author"] = spider_data.author
  133.         data["create_time"] = ""
  134.  
  135.         print(data)
  136.         # logger.info(data["title"])
  137.  
  138.         mongodao.search_upsert_one(filter={"url": data["url"]},
  139.                               data=data,
  140.                               ids={"update_time": 0,
  141.                                    "D_id": 0})
  142.

 插入数据

  1. class KsRoomInfo(Base):
  2.     __tablename__ = 'kuaishou_room_info'
  3.     sys_id = Column(Integer, primary_key=True, autoincrement=True)
  4.     roomId = Column(String(64))  # 游戏ID
  5.     cityName = Column(String(64))
  6.     nickName = Column(String(64))
  7.     desc = Column(String(128))  # 比赛ID
  8.     fans = Column(String(64))
  9.     follow = Column(String(64))
  10.     video_count = Column(String(64))
  11.     game_caty = Column(String(64))
  12.     user_caty = Column(String(64))
  13.     sys_time = Column(DateTime, default=datetime.datetime.now)
  14.     sys_timeindex = Column(Integer)
  15.  
  16.  
  17.  
  18. def saveData(datas):
  19.     session=sessionmaker(ENGINE2)()
  20.     session.add(datas)
  21.     session.commit()
  22.  
  23.  
  24. ks = KsRoomInfo(
  25.                         # roomId=data["data"]["sensitiveUserInfo"]["kwaiId"],
  26.                         roomId=room["args1"],
  27.                         cityName=data["data"]["sensitiveUserInfo"]["cityName"],
  28.                         nickName=data2["nickName"] if data2["nickName"] else room["nickName"],
  29.                         fans=data["data"]["sensitiveUserInfo"]["counts"]["fan"],
  30.                         follow=data["data"]["sensitiveUserInfo"]["counts"]["follow"],
  31.                         video_count=data["data"]["sensitiveUserInfo"]["counts"]["photo"],
  32.                         desc=data2["desc"] if data2["desc"] else room["nickName"],
  33.                         sys_time=sys_time,
  34.                         sys_timeindex=sys_timeindex,
  35.                         game_caty=room["game_caty"],
  36.                         user_caty=room["user_caty"],
  37.                         )
  38.  
  39. saveData(ks)
  1. # -*- coding:utf8 -*-
  2. # time: 19-1-18 下午3:56
  3. # Author:zhuhao
  4. # ­*­ coding: utf­8 ­*­
  5. """
  6. """
  7. import json
  8. import os
  9. from io import BytesIO
  10. import sys
  11.  
  12. import bson
  13. from pymongo import MongoClient, ReturnDocument
  14.  
  15.  
  16. # from utils.constants import DataStatus
  17. url = "localhost"
  18. db = "collet"
  19. coll_name = "collect_notice"
  20. class MongoDao(object):
  21.     def __init__(self,url,db,coll_name):
  22.         self.collection = self._get_coll(url=url, db=db, coll_name=coll_name)
  23.  
  24.     def _get_coll(self,url, db, coll_name):
  25.         db = _get_db(url, db)
  26.         return db[coll_name]
  27.  
  28.     def _get_connection(self,url):
  29.         return MongoClient(url)
  30.  
  31.     def _get_db(self,url, db):
  32.         client = _get_connection(url)
  33.         return client[db]
  34.  
  35.     def search_one(self, filter={}):
  36.         return self.collection.find_one(filter, no_cursor_timeout=True)
  37.  
  38.     def search(self, filter={}):
  39.         return self.collection.find(filter, no_cursor_timeout=True)
  40.  
  41.     def insert_one(self,data={}):
  42.         self.collection.insert_one(document=data)
  43.  
  44.     def update_one(self, filter={}, data={}, upsert=False):
  45.         """
  46.         upsert = update + insert
  47.         :param filter:
  48.         :param data:
  49.         :param upsert:
  50.         :param collection:
  51.         :return:
  52.         """
  53.         self.collection.update_one(filter=filter, update={"$set": data}, upsert=upsert)
  54.  
  55.     def search_upsert_one(self, filter={}, data={}, ids={}):
  56.         """
  57.         :param filter:
  58.         :param data:
  59.         :param ids: {"key1":1,"key2":0} 合并(1)or覆盖(0)
  60.         :return:
  61.         """
  62.         if filter is not {}:
  63.             record = self.collection.find_one(filter=filter)
  64.         else:
  65.             record = None
  66.         if record:
  67.             for id_key,v in ids.items():
  68.                 if id_key in record:
  69.                     if v == 1 and isinstance(data[id_key],record[id_key]):
  70.                         data[id_key] += record[id_key]
  71.                     else:
  72.                         data[id_key] = record[id_key]
  73.             self.collection.update_one(filter=filter, update={"$set": data})
  74.         else:
  75.             # 插入
  76.             self.collection.insert_one(document=data)
  77.  
  78.  
  79.  
  80.  
  81.  
  82.  
  83. def insertFile(file_path):
  84.     """
  85.     insert file whose size < 16M into mongodb
  86.     :param file_path:
  87.     :return:
  88.     """
  89.     try:
  90.         fin = open(file_path, "rb")
  91.         img = fin.read()
  92.         data = BytesIO(img)
  93.         fin.close()
  94.         # conn = MongoClient('10.101.12.23', 27017)
  95.         conn = MongoClient('101.132.167.173', 8003)
  96.         db = conn.mydb
  97.         coll = db.test_set
  98.         with open(file_path, 'rb') as myimage:
  99.             content = BytesIO(myimage.read())
  100.             coll.save(dict(
  101.                 content=bson.binary.Binary(content.getvalue()),
  102.                 filename=file_path.split(".")[0],
  103.                 file_format=file_path.split(".")[1]
  104.             ))
  105.     except Exception as e:
  106.         print(e)
  107.         sys.exit(1)
  108.     # finally:
  109.     #     imgput.close()  todo
  110.  
  111.  
  112. def file_read_binary(plugin_name, file_path=None):
  113.     """
  114.     从mongo中读文件，文件为二进制
  115.     :return:
  116.     """
  117.     try:
  118.         # fin = open(os.path.join(file_path,plugin_name), "wb")
  119.         # img = fin.read()
  120.         # data = StringIO(img)
  121.         # fin.close()
  122.         conn = MongoClient('101.132.167.173', 8003)
  123.         db = conn.mydb
  124.         files = db.test_set
  125.         res = files.find_one({"filename": plugin_name})
  126.         fin = open(os.path.join(file_path, plugin_name) + "." + res["file_format"], "wb")
  127.         print(res)
  128.         content = res['content']
  129.  
  130.         fin.write(content)
  131.         # print('sum:', files.count())
  132.         # insertimg = imgput.put(data)
  133.         # print insertimg
  134.     except Exception as e:
  135.         print(e)
  136.         sys.exit(1)
  137.  
  138.  
  139. def _get_connection(url):
  140.     return MongoClient(url)
  141.  
  142.  
  143. def _get_db(url, db):
  144.     client = _get_connection(url)
  145.     return client[db]
  146.  
  147.  
  148. def get_coll(url, db, coll_name):
  149.     db = _get_db(url, db)
  150.     return db[coll_name]
  151.  
  152.  
  153. def search(url, db, coll_name, filter={}):
  154.     collection = get_coll(url=url, db=db, coll_name=coll_name)
  155.     return collection.find(filter, no_cursor_timeout=True)
  156.  
  157.  
  158. def search_one(url, db, coll_name, filter={}, collection=None):
  159.     if collection is None:
  160.         collection = get_coll(url=url, db=db, coll_name=coll_name)
  161.     return collection.find_one(filter, no_cursor_timeout=True)
  162.  
  163.  
  164. def fetch_one(url, db, coll_name, filter={}, ):
  165.     collection = get_coll(url=url, db=db, coll_name=coll_name)
  166.     # return collection.find_one(filter)
  167.     return collection.find_one_and_update(filter,
  168.                                           {"$set": {"status": "1"}},
  169.                                           return_document=ReturnDocument.AFTER)
  170.  
  171.  
  172. def get_collection_names(url):
  173.     """
  174.     获取mongodb所有数据库对应的collection名
  175.     :param url:
  176.     :return:
  177.     """
  178.     client = _get_connection(url)
  179.     d = dict((db, [collection for collection in client[db].collection_names()])
  180.              for db in client.database_names())
  181.     return d
  182.  
  183.  
  184. def update_one(url, db, coll_name, filter={}, data={}, upsert=False,collection=None):
  185.     if collection is None:
  186.         collection = get_coll(url=url, db=db, coll_name=coll_name)
  187.     else:
  188.         collection.update_one(filter=filter, update={"$set": data}, upsert=upsert)
  189.     # db.stu.update_one({"age": 18}, {"$set": {"age": 100}})
  190.  
  191.  
  192. def insert_one(url, db, coll_name, data={}):
  193.     collection = get_coll(url=url, db=db, coll_name=coll_name)
  194.     collection.insert_one(document=data)
  195.  
  196.  
  197. def insert_or_update_one(url, db, coll_name, filter={}, data={},id_key=None):
  198.     collection = get_coll(url=url, db=db, coll_name=coll_name)
  199.     if filter is not {}:
  200.         record = collection.find_one(filter=filter)
  201.     else:
  202.         record = None
  203.     if record:
  204.         # 更新
  205.         if "update_time" in record:
  206.             data["update_time"] = record["update_time"]
  207.         if id_key is not None:
  208.             data[id_key] = record[id_key]
  209.         collection.update_one(filter=filter, update={"$set": data})
  210.     else:
  211.         # 插入
  212.         collection.insert_one(document=data)
  213.  
  214.  
  215. def save_or_update_one(url, db, coll_name, filter={}, data={}):
  216.     collection = get_coll(url=url, db=db, coll_name=coll_name)
  217.     record = collection.find_one(filter=filter)
  218.     if record:
  219.         # 更新
  220.         data["D_id"] = record["D_id"]
  221.         collection.update_one(filter=filter, update={"$set": data})
  222.     else:
  223.         # 插入
  224.         collection.insert_one(document=data)
  225.  
  226.  
  227. if __name__ == "__main__":
  228.     # print (search(url = 'mongodb://101.132.167.173:8013/',db="mydb",coll_name="test_clean")[0]["result"])
  229.     print(get_collection_names(url='mongodb://101.132.167.173:8014/')["pyspider_resultdb1"])
  230.     # result = fetch_one(url='mongodb://101.132.167.173:8014/',
  231.     #              db="mydb",
  232.     #              coll_name="test_clean",
  233.     #              filter={'status': {'$exists': 1}})
  234.     # print(result)
  235.     # if result:
  236.     #     print("00")
  237.     # else:
  238.     #     print("11")

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/菜鸟追梦旅行/article/detail/613184
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号