PyMongo 기본기와 Atlas 연동 - Mongodb와 Python (상)

PyMongo Basics and Atlas Connection - MongoDB and Python (Part 1)

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, BulkWriteError
from bson.objectid import ObjectId
import datetime
 
  # --- MongoDB Atlas 연결 정보 ---
  # 중요: 실제 사용 시에는 <username>, <password>, <cluster-url>, <dbname>을
  #       본인의 Atlas 클러스터 정보로 반드시 수정해야 합니다.
  #       보안을 위해 환경 변수나 설정 파일을 사용하는 것이 좋습니다.
ATLAS_URI = "mongodb+srv://<username>:<password>@<cluster-url>/<dbname>?retryWrites=true&w=majority"
DATABASE_NAME = "<dbname>" # Atlas 연결 문자열의 dbname과 동일하게, 또는 사용할 DB 이름
COLLECTION_NAME = "users_python_외전" # 테스트용 컬렉션 이름
 
def main():
  client = None # finally 블록에서 사용하기 위해 외부에 선언
  try:
    # --- 1. MongoDB Atlas 연결 ---
    print("MongoDB Atlas에 연결을 시도합니다...")
    client = MongoClient(ATLAS_URI, serverSelectionTimeoutMS=5000) # 5초 타임아웃 설정
    client.admin.command('ping') # 연결 상태 확인
    print("MongoDB Atlas에 성공적으로 연결되었습니다!\n")
 
    db = client[DATABASE_NAME]
    users_collection = db[COLLECTION_NAME]
 
    # 테스트를 위해 시작 시 컬렉션 내용 비우기 (선택 사항)
    print(f"'{COLLECTION_NAME}' 컬렉션의 기존 문서를 삭제합니다...")
    users_collection.delete_many({})
    print("기존 문서 삭제 완료.\n")
 
    # --- 2. 데이터 생성 (Create) 심화 ---
    print("--- 데이터 생성 (Create) ---")
    # 2.1 insert_one
    new_user_doc_1 = {
      "name": "유관순",
      "age": 18,
      "email": "yoo@example.com",
      "tags": ["patriot", "activist"],
      "joined_date": datetime.datetime(2024, 1, 1, 10, 0, 0, tzinfo=datetime.timezone.utc)
    }
    result_one = users_collection.insert_one(new_user_doc_1)
    print(f"insert_one 성공, ID: {result_one.inserted_id}, 확인됨: {result_one.acknowledged}")
 
    # 2.2 insert_many
    my_specific_id_for_kimkoo = ObjectId() # 김구 선생님 _id 미리 생성
    new_users_list = [
      {"name": "안중근", "age": 30, "occupation": " 독립운동가", "country": "조선"},
      {"name": "윤봉길", "age": 24, "occupation": " 독립운동가", "_id": ObjectId(), "country": "조선"},
      {"name": "김구", "age": 50, "email": "kimkoo@example.com", "_id": my_specific_id_for_kimkoo, "country": "대한민국 임시정부"}
    ]
    try:
      result_many = users_collection.insert_many(new_users_list, ordered=False)
      print(f"insert_many 성공, ID들: {result_many.inserted_ids}\n")
    except BulkWriteError as bwe:
      print(f"insert_many 중 대량 쓰기 오류 발생: {bwe.details}\n")
 
 
    # --- 3. 데이터 조회 (Read) 심화 ---
    print("--- 데이터 조회 (Read) ---")
    # 3.1 find_one (특정 조건 + 정렬 + 프로젝션)
    oldest_patriot = users_collection.find_one(
      {"occupation": " 독립운동가"},
      sort=[("age", -1)], # pymongo.DESCENDING 대신 -1 사용 가능
      projection={"name": 1, "age": 1, "_id": 0}
    )
    if oldest_patriot:
      print(f"find_one (가장 나이가 많은 독립운동가): {oldest_patriot}")
 
    # 3.2 find (특정 조건 + 프로젝션 + 정렬 + limit)
    korean_patriots = users_collection.find(
      {"country": {"$in": ["조선", "대한민국 임시정부"]}}, # 조선 또는 대한민국 임시정부
      projection={"name": 1, "age": 1, "country": 1, "_id": 0},
      limit=5,
      sort=[("name", 1)] # pymongo.ASCENDING 대신 1 사용 가능
    )
    print("find (조선/대한민국 임시정부 소속 독립운동가 최대 5명, 이름순):")
    for patriot in korean_patriots:
      print(patriot)
    print("")
 
 
    # --- 4. 데이터 수정 (Update) 심화 ---
    print("--- 데이터 수정 (Update) ---")
    # 4.1 update_one (upsert 예시)
    update_yoo_result = users_collection.update_one(
      {"name": "유관순"},
      {"$set": {"age": 19, "status": "martyr"}, "$currentDate": {"lastModified": True}},
      upsert=True
    )
    print(f"update_one ('유관순'): 일치: {update_yoo_result.matched_count}, 수정: {update_yoo_result.modified_count}, 업서트ID: {update_yoo_result.upserted_id}")
 
    # 4.2 update_many
    update_country_result = users_collection.update_many(
      {"occupation": " 독립운동가"},
      {"$set": {"country": "대한민국 독립유공자"}, "$inc": {"recognition_level": 1}}
    )
    print(f"update_many ('독립운동가' 국가 및 등급 수정): {update_country_result.modified_count}명 수정됨.\n")
 
    # 수정된 김구 선생님 정보 확인
    kimkoo_updated = users_collection.find_one({"_id": my_specific_id_for_kimkoo})
    print(f"업데이트된 김구 선생님 정보: {kimkoo_updated}\n")
 
    # --- 5. 데이터 삭제 (Delete) 심화 ---
    print("--- 데이터 삭제 (Delete) ---")
    # 5.1 delete_one
    # 가장 어린 독립운동가 한 명 삭제 (테스트용)
    youngest_fighter = users_collection.find_one({"occupation": "대한민국 독립유공자"}, sort=[("age", 1)])
    if youngest_fighter:
      delete_one_result = users_collection.delete_one({"_id": youngest_fighter["_id"]})
      print(f"delete_one (가장 어린 독립유공자 '{youngest_fighter['name']}' 삭제): {delete_one_result.deleted_count}명 삭제됨.")
    else:
      print("삭제할 가장 어린 독립유공자가 없습니다.")
    
    # 5.2 delete_many (특정 조건 - 주의해서 사용)
    # 예시: 'status' 필드가 없는 문서 모두 삭제 (여기서는 실행하지 않음)
    # delete_many_result = users_collection.delete_many({"status": {"$exists": False}})
    # print(f"delete_many (status 필드 없는 문서 삭제): {delete_many_result.deleted_count}명 삭제됨.")
    print("delete_many 예시는 주석 처리됨 (데이터 보호를 위해).\n")
 
 
    # --- 6. BSON 데이터 타입 확인 예시 ---
    print("--- BSON 데이터 타입 확인 ---")
    doc_to_check = users_collection.find_one({"name": "유관순"})
    if doc_to_check:
      print(f"'유관순' 문서의 _id 타입: {type(doc_to_check['_id'])}")
      print(f"'유관순' 문서의 joined_date 타입: {type(doc_to_check['joined_date'])}")
      print(f"'유관순' 문서의 lastModified 타입: {type(doc_to_check.get('lastModified'))}") # 없을 수 있으므로 .get() 사용
    print("")
 
  except ConnectionFailure as e:
    print(f"Atlas 연결 실패: {e}")
  except BulkWriteError as bwe: # insert_many 등에서 발생 가능
    print(f"대량 쓰기 작업 중 오류 발생: {bwe.details}")
  except Exception as e:
    print(f"알 수 없는 오류가 발생했습니다: {e}")
    # 스택 트레이스 확인을 위해 다음 줄 주석 해제 가능
    # import traceback
    # traceback.print_exc()
  finally:
    if client:
      client.close()
      print("MongoDB 연결이 해제되었습니다.")
 
if __name__ == "__main__":
  main()

MongoDB Atlas에 연결을 시도합니다...
MongoDB Atlas에 성공적으로 연결되었습니다!
 
'users_python_외전' 컬렉션의 기존 문서를 삭제합니다...
기존 문서 삭제 완료.
 
--- 데이터 생성 (Create) ---
insert_one 성공, ID: 681c42b23ae26e5f6876b7c2, 확인됨: True
insert_many 성공, ID들: [ObjectId('681c42b23ae26e5f6876b7c5'), ObjectId('681c42b23ae26e5f6876b7c4'), ObjectId('681c42b23ae26e5f6876b7c3')]
 
--- 데이터 조회 (Read) ---
find_one (가장 나이가 많은 독립운동가): {'name': '안중근', 'age': 30}
find (조선/대한민국 임시정부 소속 독립운동가 최대 5명, 이름순):
{'name': '김구', 'age': 50, 'country': '대한민국 임시정부'}
{'name': '안중근', 'age': 30, 'country': '조선'}
{'name': '윤봉길', 'age': 24, 'country': '조선'}
 
--- 데이터 수정 (Update) ---
update_one ('유관순'): 일치: 1, 수정: 1, 업서트ID: None
update_many ('독립운동가' 국가 및 등급 수정): 2명 수정됨.
 
업데이트된 김구 선생님 정보: {'_id': ObjectId('681c42b23ae26e5f6876b7c3'), 'name': '김구', 'age': 50, 'email': 'kimkoo@example.com', 'country': '대한민국 임시정부'}
 
--- 데이터 삭제 (Delete) ---
삭제할 가장 어린 독립유공자가 없습니다.
delete_many 예시는 주석 처리됨 (데이터 보호를 위해).
 
--- BSON 데이터 타입 확인 ---
'유관순' 문서의 _id 타입: <class 'bson.objectid.ObjectId'>
'유관순' 문서의 joined_date 타입: <class 'datetime.datetime'>
'유관순' 문서의 lastModified 타입: <class 'datetime.datetime'>
 
MongoDB 연결이 해제되었습니다.