Socket服务端交互时查询重复与多数据库生成问题求助

问题概述 #

通过三个独立Python文件实现Socket服务端、客户端及验证数据库，需求是客户端通过服务端查询DNI（社会号码）是否存在，但存在两个问题：

1. 查询新DNI时返回旧结果，多次查询后旧结果重复出现
2. 不同文件夹下生成多个相同的personas.db数据库文件

原代码展示 #

数据库文件（personas.py） #

from sqlalchemy import create_engine, Column, Integer, String, select
from sqlalchemy.orm import sessionmaker, declarative_base
class Personas():
    personas = create_engine('sqlite:///personas.db')
    Session = sessionmaker(bind=personas)
    sessionDB = Session()
    Base = declarative_base()

class TablePersona(Personas.Base):
    __tablename__ = 'Personas'
    id = Column('id', Integer, primary_key=True)
    name = Column('name', String)
    first_name = Column('firstName', String)
    dni = Column('DNI', Integer)

    def __init__(self, first_name, name, dni):
        self.name = name
        self.first_name = first_name
        self.dni = dni

    def __repr__(self):
        return f'{self.apellido} {self.nombre} tiene DNI={self.dni}'

def vizualizar_por_dni(param):
    dni = Personas.sessionDB.query(TablePersona).filter(TablePersona.dni == param).all()
    res = dni
    if len(res) > 0:
        for i in res:
            return i
    else:
        return f'Persona con DNI: {param} no existe!'

def insertar(nombre, apellido, dni):
    persona = TablePersona(nombre, apellido, dni)
    Personas.sessionDB.add(persona)
    Personas.sessionDB.commit()

def main():
    Personas.Base.metadata.create_all(Personas.personas)

Socket服务端文件（socket_server.py） #

import socket

from personas import vizualizar_por_dni

def server():
    # he creado server con "with" loop para olvider de ese .close()
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        server_address = (socket.gethostbyname(socket.gethostname()), 1234)
        s.bind(server_address)
        s.listen(10)
        print("Server address is: ip: {} port: {}".format(*server_address))
        connection, address = s.accept()
        with connection:
            print(f"Cliente {address} connected")
            while True:
                data = connection.recv(10000000)
                if not data:
                    print("No he recibido datas!")
                    break
                datas_recibido = data.decode()
                message_to_send = f'{vizualizar_por_dni(datas_recibido)}'.encode()
                if len(message_to_send) > 0:
                    connection.sendall(message_to_send)
                else:
                    message_reply = b'Ningun persona con ese DNI!'
                    connection.sendall(message_reply)

Socket客户端文件（socket_client.py） #

import socket

def client():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((socket.gethostname(), 1234))
        while True:
            usuario = input("Introduce el DNI: \n")
            message = usuario.encode()
            s.sendall(message)
            if s.sendall(message):
                print("the client had sent the message")
            data = s.recv(10000000)
            received_message = data.decode()
            if len(received_message) > 0:
                print(received_message)
            else:
                print("Nothing received")
            continuar = input('Quieres que continuar? S or N')
            if continuar.capitalize() == 'S':
                continue
            elif continuar.capitalize() == 'N':
                break
        print("Connection closed")

问题解决方法 #

问题1：查询结果重复/返回旧结果 #

原因

客户端代码中重复调用了两次sendall，导致服务端接收到两次相同的DNI请求并返回两次响应；客户端recv时会将两次响应拼接在一起，表现为旧结果重复出现。

修复步骤

删除客户端中重复的sendall调用，同时优化输入空白字符处理。

问题2：多数据库文件生成 #

原因

数据库连接字符串使用相对路径sqlite:///personas.db，当从不同目录运行服务端/客户端时，SQLAlchemy会在当前运行目录创建数据库文件，导致多个文件生成。

修复步骤

使用文件绝对路径指定数据库位置，确保所有模块都指向同一个数据库文件；同时修正数据库代码中的属性名称错误。

修改后的完整代码 #

数据库文件（personas.py） #

import os
from sqlalchemy import create_engine, Column, Integer, String, select
from sqlalchemy.orm import sessionmaker, declarative_base

class Personas():
    # 获取当前文件所在目录，确保数据库仅生成在该目录下
    base_dir = os.path.dirname(os.path.abspath(__file__))
    db_path = os.path.join(base_dir, 'personas.db')
    personas = create_engine(f'sqlite:///{db_path}')
    Session = sessionmaker(bind=personas)
    sessionDB = Session()
    Base = declarative_base()

class TablePersona(Personas.Base):
    __tablename__ = 'Personas'
    id = Column('id', Integer, primary_key=True)
    name = Column('name', String)
    first_name = Column('firstName', String)
    dni = Column('DNI', Integer)

    def __init__(self, first_name, name, dni):
        self.name = name
        self.first_name = first_name
        self.dni = dni

    def __repr__(self):
        # 修正属性名称，与类定义一致
        return f'{self.name} {self.first_name} tiene DNI={self.dni}'

def vizualizar_por_dni(param):
    # 增加类型转换，避免字符串与整数匹配失败
    try:
        param = int(param)
    except ValueError:
        return 'DNI debe ser un número entero!'
    
    # 使用first()替代all()，无需循环直接返回结果
    dni_record = Personas.sessionDB.query(TablePersona).filter(TablePersona.dni == param).first()
    return dni_record if dni_record else f'Persona con DNI: {param} no existe!'

def insertar(nombre, apellido, dni):
    persona = TablePersona(nombre, apellido, dni)
    Personas.sessionDB.add(persona)
    Personas.sessionDB.commit()

def main():
    Personas.Base.metadata.create_all(Personas.personas)

if __name__ == '__main__':
    main()

Socket服务端文件（socket_server.py） #

import socket
from personas import vizualizar_por_dni

def server():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        server_address = (socket.gethostbyname(socket.gethostname()), 1234)
        s.bind(server_address)
        s.listen(10)
        print("Server address is: ip: {} port: {}".format(*server_address))
        # 增加循环，支持多客户端连接
        while True:
            connection, address = s.accept()
            with connection:
                print(f"Cliente {address} connected")
                while True:
                    # 缩小缓冲区大小，避免不必要的内存占用
                    data = connection.recv(1024)
                    if not data:
                        print(f"Cliente {address} desconectado")
                        break
                    # 去除输入中的空白字符
                    datas_recibido = data.decode().strip()
                    message_to_send = f'{vizualizar_por_dni(datas_recibido)}'.encode()
                    connection.sendall(message_to_send)

Socket客户端文件（socket_client.py） #

import socket

def client():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.connect((socket.gethostname(), 1234))
        while True:
            usuario = input("Introduce el DNI: \n").strip()
            if not usuario:
                print("DNI no puede estar vacío!")
                continue
            message = usuario.encode()
            # 仅调用一次sendall
            s.sendall(message)
            print("Mensaje enviado al servidor")
            # 匹配服务端的缓冲区大小
            data = s.recv(1024)
            received_message = data.decode()
            print(received_message if received_message else "Nada recibido")
            continuar = input('Quieres continuar? S or N: ').strip()
            if continuar.capitalize() == 'N':
                break
        print("Conexión cerrada")

内容的提问来源于stack exchange，提问作者alender