python知识图谱-py2neo实现neo4j的dao类

2024-01-08 17:33:04

python知识图谱-利用py2neo实现neo4j的dao类

py2neo的基本使用以及cypher的基本使用:https://blog.csdn.net/Akun_2217/article/details/135445147?spm=1001.2014.3001.5502

1. dao类需要实现的功能

neo4j实现dao类的基本单位就是子图,包括单个节点、单个子图,都属于子图。

  1. 数据库的创建

  2. 创建子图

  3. 查询子图

  4. 更改子图

  5. 删除子图

  6. 事务功能

    • 一次性插入多条数据时,要么都成功要么都失败

2. dao类具体设计

日志模块实现:https://blog.csdn.net/Akun_2217/article/details/135442731?spm=1001.2014.3001.5502

  1. 初始化对象进行连接:_init_
  2. 执行非查询cypher语句: execute
  3. 执行查询cypher语句: query
  4. 切换数据库: select_db
  5. 开启事务: begin_transaction
  6. 事务执行: transaction_execute
  7. 提交事务: commit_transaction
  8. 回滚事务: rollback_transaction
  9. 关闭连接 : close_connect
  10. 日志记录

3. 利用py2neo实现neo4j的dao类

  • 实际工作中使用时,将neo4j日志路径以及连接参数,卸载一个配置文件中。

  • cypher语言使用两种形式, 以match为例,其他类似:

    1. run(“match (n:boss) where n.name=‘biden’ set n.name=$newname”, **{‘newname’:‘aoguanhai’})
    2. run(“match (n:boss) where n.name=‘biden’ set n.name=$newname”, newname=‘auguanhai’)
import py2neo
from utils import logger_utils

logger = logger_utils.get_logger('./')


class Neo4jDao:

    def __init__(self, url, auth):
        try:
            self.graph_server = py2neo.GraphService(url, auth=auth)
            logger.info(f'服务连接成功,主机地址:{url}....')
        except Exception as e:
            logger.info(f'服务连接失败,原因:{e}, 主机地址:{url}')
            # 使用默认数据库
            self.graph_server = None
        self.graph = None
        self.tx = None

    def select_db(self, db_name):
        if self.graph_server:
            if db_name in self.graph_server.keys():
                self.graph = self.graph_server[db_name]
                self.tx = None
                logger.info(f'已经切换数据库,数据库切换为:{db_name}, 当前事务清零')
            else:
                logger.error(f'切换数据库失败, 数据库{db_name}不存在')
        else:
            logger.error(f'数据库切换失败,原因:数据服务未连接...')

    def execute(self, cypher, **args):
        '''自动提交-执行非查询cypher,无返回结果'''
        try:
            self.graph.run(cypher, args)
            logger.info(f'自动提价执行-{cypher}执行完成。。。')
        except Exception as e:
            if not self.graph:
                logger.error(f'自动提交执行失败-{cypher}执行失败,原因:未选择数据库')
            else:
                logger.error(f'自动提交执行失败-{cypher}执行失败,原因: {e}')

    def query(self, cypher, **args):
        try:
            cursor = self.graph.run(cypher, args)
            logger.info(f'查询成功-{cypher}执行成功')
            return cursor

        except Exception as e:
            if not self.graph:
                logger.error(f'查询执行失败-{cypher}执行失败,原因:未选择数据库')
            else:
                logger.error(f'查询执行失败-{cypher}执行失败,原因: {e}')

    def transaction_execute(self, cypher, **args):
        '''非自动提交事务执行非查询语句'''
        try:
            self.tx.run(cypher, args)
            logger.info(f'事务执行成功, {cypher}执行成功')
        except Exception as e:
            if not self.tx:
                logger.error(f'事务执行失败,{cypher}执行失败,原因:未开启事务')
            else:
                logger.error(f'事务执行失败,{cypher}执行失败,原因: {e}')

    def begin_transaction(self):
        if self.tx:
            logger.warning('开启事务冲突,当前连接已经开启事务')
            logger.warning('关闭当前连接的事务')
            self.tx = None
        try:
            self.tx = self.graph.begin()
            logger.info('事务已经开启...')
        except Exception as e:
            if not self.graph:
                logger.error(f'事务开启失败,原因:未选择数据库')
            else:
                logger.error(f'事务开启失败,原因: {e}')

    def rollback_transaction(self):
        try:
            self.graph.rollback(self.tx)
            logger.info('事务已经回滚...')
        except Exception as e:
            logger.error(f'事务回滚失败, 原因{e}')

    def commit_transaction(self):
        try:
            self.graph.commit(self.tx)
            self.tx = None
            logger.info('事务提交已经完成...事务结束')
        except Exception as e:
            logger.error(f'事务提交失败, 原因{e}')

    def close_connect(self):
        self.graph_server = None
        logger.info('neo4j服务连接诶已经关闭...')


if __name__ == '__main__':
    neo4j = Neo4jDao("bolt://localhost:7687", auth=("neo4j", "a86xavwahgwwy@@@"))
    neo4j.select_db('neo4j')
    record = neo4j.query('match (n) return n')
    print(record.to_data_frame())

文章来源:https://blog.csdn.net/Akun_2217/article/details/135461451
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。